Region fixes

Author: m09526

java/compaction/compaction-datafusion/src/main/java/sleeper/compaction/datafusion/DataFusionCompactionRunner.java

@@ -126,7 +126,7 @@ private static FFICommonConfig createCompactionParams(CompactionJob job, TablePr
         if (DataEngine.AGGREGATION_ITERATOR_NAME.equals(job.getIteratorClassName())) {
             params.iterator_config.set(job.getIteratorConfig());
         }
-        FFISleeperRegion partitionRegion = new FFISleeperRegion(runtime, region);
+        FFISleeperRegion partitionRegion = new FFISleeperRegion(runtime, schema, region);
         params.setRegion(partitionRegion);
         params.validate();
         return params;

java/foreign-bridge/src/main/java/sleeper/foreign/FFISleeperRegion.java

@@ -20,8 +20,10 @@
 
 import sleeper.core.range.Range;
 import sleeper.core.range.Region;
+import sleeper.core.schema.Schema;
 import sleeper.foreign.bridge.FFIArray;
 
+import java.util.List;
 import java.util.Objects;
 
 /**
@@ -46,50 +48,55 @@ public class FFISleeperRegion extends Struct {
     public final FFIArray<java.lang.Boolean> maxs_inclusive = new FFIArray<>(this);
 
     public FFISleeperRegion(jnr.ffi.Runtime runtime) {
-        this(runtime, null);
+        this(runtime, null, null);
     }
 
     /**
      * Creates and validates Sleeper region into an FFI compatible form.
      *
      * @param  runtime               FFI runtime
+     * @param  schema                the table schema
      * @param  region                the Sleeper partition region
      * @throws IllegalStateException when the region data is invalid
      */
-    public FFISleeperRegion(jnr.ffi.Runtime runtime, Region region) {
+    public FFISleeperRegion(jnr.ffi.Runtime runtime, Schema schema, Region region) {
         super(runtime);
         if (region != null) {
-            populateRegion(region);
+            populateRegion(schema, region);
         }
     }
 
     /**
      * Set the region data in this FFI object.
      *
+     * @param  schema               the table schema
      * @param  region               region data to copy into FFI object
      * @throws NullPointerException if {@code region} is {@code null}
      */
     @SuppressWarnings(value = "checkstyle:avoidNestedBlocks")
-    public void populateRegion(Region region) {
+    public void populateRegion(Schema schema, Region region) {
+        Objects.requireNonNull(schema, "schema");
         Objects.requireNonNull(region, "region");
+        List<java.lang.String> rowKeysOrdered = schema.getRowKeyFieldNames();
+        List<Range> orderedRanges = rowKeysOrdered.stream().map(rowKeyName -> region.getRange(rowKeyName)).toList();
         // Extra braces: Make sure wrong array isn't populated to wrong pointers
         {
             // This array can't contain nulls
-            Object[] regionMins = region.getRanges().stream().map(Range::getMin).toArray();
+            Object[] regionMins = orderedRanges.stream().map(Range::getMin).toArray();
             this.mins.populate(regionMins, false);
         }
         {
-            java.lang.Boolean[] regionMinInclusives = region.getRanges().stream().map(Range::isMinInclusive)
+            java.lang.Boolean[] regionMinInclusives = orderedRanges.stream().map(Range::isMinInclusive)
                     .toArray(java.lang.Boolean[]::new);
             this.mins_inclusive.populate(regionMinInclusives, false);
         }
         {
             // This array can contain nulls
-            Object[] regionMaxs = region.getRanges().stream().map(Range::getMax).toArray();
+            Object[] regionMaxs = orderedRanges.stream().map(Range::getMax).toArray();
             this.maxs.populate(regionMaxs, true);
         }
         {
-            java.lang.Boolean[] regionMaxInclusives = region.getRanges().stream().map(Range::isMaxInclusive)
+            java.lang.Boolean[] regionMaxInclusives = orderedRanges.stream().map(Range::isMaxInclusive)
                     .toArray(java.lang.Boolean[]::new);
             this.maxs_inclusive.populate(regionMaxInclusives, false);
         }

java/query/query-datafusion/src/main/java/sleeper/query/datafusion/DataFusionLeafPartitionRowRetriever.java

@@ -148,7 +148,7 @@ private static FFILeafPartitionQueryConfig createFFIQueryData(LeafPartitionQuery
         if (DataEngine.AGGREGATION_ITERATOR_NAME.equals(query.getQueryTimeIteratorClassName())) {
             common.iterator_config.set(query.getQueryTimeIteratorConfig());
         }
-        FFISleeperRegion partitionRegion = new FFISleeperRegion(runtime, query.getPartitionRegion());
+        FFISleeperRegion partitionRegion = new FFISleeperRegion(runtime, dataReadSchema, query.getPartitionRegion());
         common.setRegion(partitionRegion);
         common.validate();
 
@@ -161,7 +161,7 @@ private static FFILeafPartitionQueryConfig createFFIQueryData(LeafPartitionQuery
             queryConfig.requested_value_fields_set.set(false);
         }
 
-        FFISleeperRegion[] ffiRegions = query.getRegions().stream().map(region -> new FFISleeperRegion(runtime, region)).toArray(FFISleeperRegion[]::new);
+        FFISleeperRegion[] ffiRegions = query.getRegions().stream().map(region -> new FFISleeperRegion(runtime, dataReadSchema, region)).toArray(FFISleeperRegion[]::new);
         queryConfig.setQueryRegions(ffiRegions);
         queryConfig.write_quantile_sketch.set(false);
         queryConfig.explain_plans.set(true);