Expose sorting columns (#727)

Author: brancz

dynparquet/schema.go

@@ -650,7 +650,13 @@ func (s *Schema) Columns() []ColumnDefinition {
 	return s.columns
 }
 
-func (s *Schema) SortingColumns() []ColumnDefinition {
+func (s *Schema) SortingColumns() []SortingColumn {
+	sCols := make([]SortingColumn, len(s.sortingColumns))
+	copy(sCols, s.sortingColumns)
+	return sCols
+}
+
+func (s *Schema) ColumnDefinitionsForSortingColumns() []ColumnDefinition {
 	sCols := make([]ColumnDefinition, len(s.sortingColumns))
 	for i, col := range s.sortingColumns {
 		sCols[i] = s.columns[s.columnIndexes[col.ColumnName()]]

query/physicalplan/physicalplan.go

@@ -309,7 +309,7 @@ func Build(
 	if s != nil {
 		// TODO(asubiotto): There are cases in which the schema can be nil.
 		// Eradicate these.
-		oInfo.sortingCols = s.SortingColumns()
+		oInfo.sortingCols = s.ColumnDefinitionsForSortingColumns()
 	}
 
 	var visitErr error

table.go

@@ -1468,7 +1468,7 @@ func (t *Table) writeRecordsToParquet(w io.Writer, records []arrow.Record, sortI
 // caller should fall back to normal compaction. On success, the caller is
 // responsible for releasing the returned records.
 func (t *Table) distinctRecordsForCompaction(compact []parts.Part) ([]arrow.Record, error) {
-	sortingCols := t.schema.SortingColumns()
+	sortingCols := t.schema.ColumnDefinitionsForSortingColumns()
 	columnExprs := make([]logicalplan.Expr, 0, len(sortingCols))
 	for _, col := range sortingCols {
 		var expr logicalplan.Expr