Merge pull request #1020: [beam-core] add min-initial-splits to BatchLogRead

Author: je-ik

beam/core/src/main/java/cz/o2/proxima/beam/core/direct/io/BatchLogRead.java

@@ -250,7 +250,7 @@ public void splitRestriction(
         List<Partition> partitions = restriction.getPartitions();
         partitions.sort(Comparator.comparing(Partition::getMinTimestamp));
         int pos = 0;
-        int reduced = (int) Math.sqrt(partitions.size());
+        int reduced = Math.max(minInitialSplits, (int) Math.sqrt(partitions.size()));
         if (maxInitialSplits > 0 && reduced > maxInitialSplits) {
           reduced = maxInitialSplits;
         }
@@ -307,6 +307,7 @@ public TypeDescriptor<StreamElement> getOutputTypeDescriptor() {
   private final long startStamp;
   private final long endStamp;
   private final int maxInitialSplits;
+  private final int minInitialSplits;
 
   @VisibleForTesting
   BatchLogRead(
@@ -324,16 +325,24 @@ public TypeDescriptor<StreamElement> getOutputTypeDescriptor() {
     this.readerFactory = readerFactory;
     this.startStamp = startStamp;
     this.endStamp = endStamp;
-    this.maxInitialSplits = readInitialSplits(cfg);
+    this.maxInitialSplits = readMaxInitialSplits(cfg);
+    this.minInitialSplits = readMinInitialSplits(cfg);
   }
 
-  private int readInitialSplits(Map<String, Object> cfg) {
+  private int readMaxInitialSplits(Map<String, Object> cfg) {
     return Optional.ofNullable(cfg.get("batch.max-initial-splits"))
         .map(Object::toString)
         .map(Integer::valueOf)
         .orElse(-1);
   }
 
+  private int readMinInitialSplits(Map<String, Object> cfg) {
+    return Optional.ofNullable(cfg.get("batch.min-initial-splits"))
+        .map(Object::toString)
+        .map(Integer::valueOf)
+        .orElse(-1);
+  }
+
   @Override
   public PCollection<StreamElement> expand(PBegin input) {
     long delayMs =

beam/core/src/test/java/cz/o2/proxima/beam/core/BeamDataOperatorTest.java

@@ -450,7 +450,8 @@ public void testBatchLogReadConfigParsing() {
     beam.getBatchUpdates(p, armed);
     DataAccessor accessor = beam.getAccessorFor(repo.getFamilyByName("gateway-storage-batch"));
     Map<String, Object> cfg = ((DirectDataAccessorWrapper) accessor).getCfg();
-    assertEquals(1, cfg.get("batch.max-initial-splits"));
+    assertEquals(3, cfg.get("batch.max-initial-splits"));
+    assertEquals(2, cfg.get("batch.min-initial-splits"));
   }
 
   private void terminatePipeline(

beam/core/src/test/resources/test-config-parsing.conf

@@ -1,7 +1,8 @@
 attributeFamilies {
   gateway-storage-batch {
     batch {
-      max-initial-splits: 1
+      max-initial-splits: 3
+      min-initial-splits: 2
     }
   }
 }