Add unit test for split boundary logic

presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java

@@ -34,6 +34,7 @@
 import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.testing.Assertions.assertContains;
 import static io.airlift.units.DataSize.Unit.MEGABYTE;
+import static io.prestosql.plugin.hive.HiveSessionProperties.getMaxInitialSplitSize;
 import static io.prestosql.plugin.hive.HiveTestUtils.SESSION;
 import static io.prestosql.spi.connector.NotPartitionedPartitionHandle.NOT_PARTITIONED;
 import static java.lang.Math.toIntExact;
@@ -77,6 +78,34 @@ public void testOutstandingSplitCount()
         assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 0);
     }
 
+    @Test
+    public void testEvenlySizedSplitRemainder()
+    {
+        DataSize initialSplitSize = getMaxInitialSplitSize(SESSION);
+        HiveSplitSource hiveSplitSource = HiveSplitSource.allAtOnce(
+                SESSION,
+                "database",
+                "table",
+                10,
+                10,
+                DataSize.of(1, MEGABYTE),
+                Integer.MAX_VALUE,
+                new TestingHiveSplitLoader(),
+                Executors.newSingleThreadExecutor(),
+                new CounterStat());
+
+        // One byte larger than the initial split max size
+        DataSize fileSize = DataSize.ofBytes(initialSplitSize.toBytes() + 1);
+        long halfOfSize = fileSize.toBytes() / 2;
+        hiveSplitSource.addToQueue(new TestSplit(1, OptionalInt.empty(), fileSize));
+
+        HiveSplit first = (HiveSplit) getSplits(hiveSplitSource, 1).get(0);
+        assertEquals(first.getLength(), halfOfSize);
+
+        HiveSplit second = (HiveSplit) getSplits(hiveSplitSource, 1).get(0);
+        assertEquals(second.getLength(), fileSize.toBytes() - halfOfSize);
+    }
+
     @Test
     public void testFail()
     {
@@ -286,17 +315,22 @@ private TestSplit(int id)
         }
 
         private TestSplit(int id, OptionalInt bucketNumber)
+        {
+            this(id, bucketNumber, DataSize.ofBytes(100));
+        }
+
+        private TestSplit(int id, OptionalInt bucketNumber, DataSize fileSize)
         {
             super(
                     "partition-name",
                     "path",
                     0,
-                    100,
-                    100,
+                    fileSize.toBytes(),
+                    fileSize.toBytes(),
                     Instant.now().toEpochMilli(),
                     properties("id", String.valueOf(id)),
                     ImmutableList.of(),
-                    ImmutableList.of(new InternalHiveBlock(0, 100, ImmutableList.of())),
+                    ImmutableList.of(new InternalHiveBlock(0, fileSize.toBytes(), ImmutableList.of())),
                     bucketNumber,
                     true,
                     false,