Issue 21 stream cuts

build.gradle

@@ -29,6 +29,7 @@ dependencies {
     compile "com.facebook.airlift:bootstrap:${airliftBootstrapVersion}"
     compile "com.facebook.airlift:json:${airliftJsonVersion}"
     compile "com.facebook.airlift:log:${airliftLogVersion}"
+    compile "com.facebook.airlift:concurrent:${airliftConcurrentVersion}"
     compile "com.facebook.airlift:configuration:${airliftConfigurationVersion}"
     compile "com.google.guava:guava:${guavaVersion}"
     compile "com.google.inject:guice:${guiceVersion}"

gradle.properties

@@ -9,6 +9,7 @@
 #
 
 airliftBootstrapVersion=0.191
+airliftConcurrentVersion=0.191
 airliftConfigurationVersion=0.191
 airliftJsonVersion=0.191
 airliftLogVersion=0.191
@@ -24,7 +25,7 @@ guavaVersion=26.0-jre
 guiceVersion=4.2.2
 jacksonVersion=2.10.0
 javaxValidationVersion=1.1.0.Final
-pravegaVersion=0.9.0
+pravegaVersion=0.10.0-2833.156e2c5-SNAPSHOT
 pravegaSchemaRegistryVersion=0.2.0
 prestoVersion=0.247
 protobufVersion=3.11.4

src/main/java/com/facebook/presto/pravega/PravegaProperties.java

@@ -24,13 +24,9 @@
 
 public class PravegaProperties
 {
-    private static final String SESSION_READER_TYPE = "reader_type";
-
     private static final String SESSION_CURSOR_DELIM_CHAR = "cursor_delim_char";
 
-    private static final String SESSION_GROUPED_EVENT_SPLITS = "grouped_event_splits";
-
-    private static final String SESSION_EVENT_READ_TIMEOUT_MS = "event_read_timeout_ms";
+    private static final String SEGMENT_RANGE_SPLIT_SIZE_BYTES = "segment_range_split_size_bytes";
 
     private final ConnectorSession session;
 
@@ -47,28 +43,14 @@ public static List<PropertyMetadata<?>> buildSessionProperties()
                 PropertyMetadata.stringProperty(
                         SESSION_CURSOR_DELIM_CHAR,
                         "character used as field separator for delimited formats",
-                        ",",
-                        false));
-
-        propertyMetadataList.add(
-                PropertyMetadata.stringProperty(
-                        SESSION_READER_TYPE,
-                        "reader type [event|grouped_event|segment_range|segment_range_per_split]",
-                        "segment_range_per_split",
-                        false));
-
-        propertyMetadataList.add(
-                PropertyMetadata.integerProperty(
-                        SESSION_GROUPED_EVENT_SPLITS,
-                        "number of splits when using grouped readers",
-                        63,
+                        "|",
                         false));
 
         propertyMetadataList.add(
                 PropertyMetadata.integerProperty(
-                        SESSION_EVENT_READ_TIMEOUT_MS,
-                        "timeout in ms to readNextEvent()",
-                        10000,
+                        SEGMENT_RANGE_SPLIT_SIZE_BYTES,
+                        "desired split size for segment range. cannot guarantee the size, it is approximate",
+                        32 * 1048576,
                         false));
 
         return propertyMetadataList;
@@ -79,18 +61,8 @@ public String getCursorDelimChar()
         return session.getProperty(SESSION_CURSOR_DELIM_CHAR, String.class);
     }
 
-    public String getReaderType()
-    {
-        return session.getProperty(SESSION_READER_TYPE, String.class);
-    }
-
-    public int getGroupedEventSplits()
-    {
-        return session.getProperty(SESSION_GROUPED_EVENT_SPLITS, Integer.class);
-    }
-
-    public int getEventReadTimeoutMs()
+    public int getSegmentRangeSplitSizeBytes()
     {
-        return session.getProperty(SESSION_EVENT_READ_TIMEOUT_MS, Integer.class);
+        return session.getProperty(SEGMENT_RANGE_SPLIT_SIZE_BYTES, Integer.class);
     }
 }

src/main/java/com/facebook/presto/pravega/PravegaRecordSet.java

@@ -23,6 +23,7 @@
 import com.facebook.presto.pravega.decoder.EventDecoder;
 import com.facebook.presto.spi.RecordCursor;
 import com.facebook.presto.spi.RecordSet;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import io.pravega.client.batch.SegmentRange;
 
@@ -81,26 +82,9 @@ public List<Type> getColumnTypes()
     @Override
     public RecordCursor cursor()
     {
-        Iterator<DecodableEvent> eventIterator;
-
-        switch (split.getReaderType()) {
-            case EVENT_STREAM:
-            case SINGLE_GROUP_EVENT_STREAM:
-                eventIterator = new EventStreamIterator(segmentManager, deserialize(split.getReaderArgs(), ReaderArgs.class), properties);
-                break;
-
-            case SEGMENT_RANGE:
-                eventIterator = new SegmentRangeIterator(segmentManager, deserialize(split.getReaderArgs(), ReaderArgs.class));
-                break;
-
-            case SEGMENT_RANGE_PER_SPLIT:
-                eventIterator = new SegmentEventIterator(segmentManager, deserialize(split.getReaderArgs(), SegmentRange.class));
-                break;
-
-            default:
-                throw new IllegalArgumentException("readerType " + split.getReaderType());
-        }
-
+        Preconditions.checkArgument(split.getReaderType() == ReaderType.SEGMENT_RANGE_PER_SPLIT);
+        Iterator<DecodableEvent> eventIterator =
+                new SegmentRangeIterator(segmentManager, deserialize(split.getReaderArgs(), SegmentRange.class));
         return new PravegaRecordCursor(eventIterator, columnHandles, eventDecoder, properties, split.getSchema().get(0).getFormat());
     }
 }

src/main/java/com/facebook/presto/pravega/PravegaSegmentManager.java

@@ -76,7 +76,7 @@ public PravegaSegmentManager(PravegaConnectorConfig connectorConfig)
         registryClient = SchemaRegistryClientFactory.withDefaultNamespace(registryConfig);
     }
 
-    private BatchClientFactory batchClientFactory(String scope)
+    BatchClientFactory batchClientFactory(String scope)
     {
         BatchClientFactory batchClientFactory = clientFactoryMap.get(scope);
         if (batchClientFactory == null) {
@@ -93,6 +93,11 @@ private BatchClientFactory batchClientFactory(String scope)
         return batchClientFactory;
     }
 
+    public StreamManager getStreamManager()
+    {
+        return StreamManager.create(clientConfig.getControllerURI());
+    }
+
     /**
      * Returns a list of {@link SegmentRange} instances.
      *