separated implementation classes into their own package

Author: zinal

src/main/java/tech/ydb/spark/connector/YdbCatalog.java

@@ -32,6 +32,10 @@
 import tech.ydb.scheme.SchemeClient;
 import tech.ydb.scheme.description.DescribePathResult;
 import tech.ydb.scheme.description.ListDirectoryResult;
+import tech.ydb.spark.connector.impl.YdbAlterTable;
+import tech.ydb.spark.connector.impl.YdbConnector;
+import tech.ydb.spark.connector.impl.YdbCreateTable;
+import tech.ydb.spark.connector.impl.YdbRegistry;
 import tech.ydb.table.SessionRetryContext;
 import tech.ydb.table.description.TableDescription;
 import tech.ydb.table.description.TableIndex;

src/main/java/tech/ydb/spark/connector/YdbExtensions.java

@@ -0,0 +1,18 @@
+package tech.ydb.spark.connector;
+
+import org.apache.spark.sql.SparkSessionExtensions;
+import org.apache.spark.sql.SparkSessionExtensionsProvider;
+import scala.runtime.BoxedUnit;
+
+/**
+ *
+ * @author zinal
+ */
+public class YdbExtensions implements SparkSessionExtensionsProvider {
+
+    @Override
+    public BoxedUnit apply(SparkSessionExtensions ext) {
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
+}

src/main/java/tech/ydb/spark/connector/YdbOptions.java

@@ -167,7 +167,7 @@ public abstract class YdbOptions {
      * @param referenced properties for the connection to be found or created
      * @return true, if properties connectionMatches, false otherwise
      */
-    public static boolean connectionMatches(Map<String, String> existing, Map<String, String> referenced) {
+    public static boolean optionsMatches(Map<String, String> existing, Map<String, String> referenced) {
         for (String propName : CONN_IDENTITY) {
             String v1 = existing.get(propName);
             String v2 = referenced.get(propName);

src/main/java/tech/ydb/spark/connector/YdbPartitionReader.java

@@ -1,17 +1,24 @@
 package tech.ydb.spark.connector;
 
+import java.io.IOException;
+
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.read.InputPartition;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.sql.connector.read.PartitionReaderFactory;
 
+import tech.ydb.spark.connector.impl.YdbScanImpl;
+
 /**
  * Partition reader factory delivers the scan options to partition reader instances.
  *
  * @author zinal
  */
 public class YdbPartitionReaderFactory implements PartitionReaderFactory {
 
+    private static final org.slf4j.Logger LOG =
+            org.slf4j.LoggerFactory.getLogger(YdbPartitionReaderFactory.class);
+
     private static final long serialVersionUID = 1L;
 
     private final YdbScanOptions options;
@@ -22,6 +29,46 @@ public YdbPartitionReaderFactory(YdbScanOptions options) {
 
     @Override
     public PartitionReader<InternalRow> createReader(InputPartition partition) {
-        return new YdbPartitionReader(options, (YdbTablePartition) partition);
+        return new YdbReader(options, (YdbTablePartition) partition);
+    }
+
+    static class YdbReader implements PartitionReader<InternalRow> {
+
+        private final YdbScanOptions options;
+        private final YdbTablePartition partition;
+        private YdbScanImpl scan;
+
+        YdbReader(YdbScanOptions options, YdbTablePartition partition) {
+            this.options = options;
+            this.partition = partition;
+        }
+
+        @Override
+        public boolean next() throws IOException {
+            if (scan == null) {
+                LOG.debug("Preparing scan for table {} at partition {}",
+                        options.getTablePath(), partition);
+                scan = new YdbScanImpl(options, partition.getRange());
+                scan.prepare();
+                LOG.debug("Scan prepared, ready to fetch...");
+            }
+            return scan.next();
+        }
+
+        @Override
+        public InternalRow get() {
+            return scan.get();
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (scan != null) {
+                LOG.debug("Closing the scan.");
+                scan.close();
+            }
+            scan = null;
+        }
+
     }
+
 }

src/main/java/tech/ydb/spark/connector/YdbPartitionReaderFactory.java

@@ -9,6 +9,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import org.apache.spark.sql.connector.catalog.SupportsDelete;
 import org.apache.spark.sql.connector.catalog.SupportsRead;
 import org.apache.spark.sql.connector.catalog.SupportsWrite;
 import org.apache.spark.sql.connector.catalog.Table;
@@ -18,12 +19,14 @@
 import org.apache.spark.sql.connector.read.ScanBuilder;
 import org.apache.spark.sql.connector.write.LogicalWriteInfo;
 import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.sources.Filter;
 import org.apache.spark.sql.types.DataType;
 import org.apache.spark.sql.types.Metadata;
 import org.apache.spark.sql.types.StructField;
 import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
+import tech.ydb.spark.connector.impl.YdbConnector;
 import tech.ydb.table.description.KeyRange;
 import tech.ydb.table.description.TableColumn;
 import tech.ydb.table.description.TableDescription;
@@ -35,7 +38,7 @@
  *
  * @author zinal
  */
-public class YdbTable implements Table, SupportsRead, SupportsWrite {
+public class YdbTable implements Table, SupportsRead, SupportsWrite, SupportsDelete {
 
     private static final org.slf4j.Logger LOG
             = org.slf4j.LoggerFactory.getLogger(YdbTable.class);
@@ -270,6 +273,24 @@ public WriteBuilder newWriteBuilder(LogicalWriteInfo info) {
         return new YdbWriteBuilder(this, info);
     }
 
+    @Override
+    public boolean canDeleteWhere(Filter[] filters) {
+        // We prefer per-row operations
+        return false;
+    }
+
+    @Override
+    public void deleteWhere(Filter[] filters) {
+        // Should not be called, as canDeleteWhere() returns false.
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
+    @Override
+    public boolean truncateTable() {
+        // TODO: implementation
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
     final YdbConnector getConnector() {
         return connector;
     }

src/main/java/tech/ydb/spark/connector/YdbTable.java

@@ -14,6 +14,8 @@
 import tech.ydb.core.Result;
 import tech.ydb.core.Status;
 import tech.ydb.core.StatusCode;
+import tech.ydb.spark.connector.impl.YdbConnector;
+import tech.ydb.spark.connector.impl.YdbRegistry;
 import tech.ydb.table.description.TableDescription;
 import tech.ydb.table.description.TableIndex;
 import tech.ydb.table.settings.DescribeTableSettings;

src/main/java/tech/ydb/spark/connector/YdbTableProvider.java

@@ -12,12 +12,15 @@
 import org.apache.spark.sql.connector.write.WriteBuilder;
 import org.apache.spark.sql.connector.write.WriterCommitMessage;
 
+import tech.ydb.spark.connector.impl.YdbWriterImpl;
+
 /**
  * YDB table writer: orchestration and partition writer factory.
  *
  * @author zinal
  */
-public class YdbWrite implements Serializable, WriteBuilder, Write, BatchWrite, DataWriterFactory {
+public class YdbWrite implements Serializable,
+        WriteBuilder, Write, BatchWrite, DataWriterFactory {
 
     private static final long serialVersionUID = 1L;
 
@@ -60,7 +63,7 @@ public void abort(WriterCommitMessage[] messages) {
 
     @Override
     public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
-        return new YdbWriterBasic(options);
+        return new YdbWriterImpl(options);
     }
 
 }

src/main/java/tech/ydb/spark/connector/YdbWrite.java

@@ -1,4 +1,4 @@
-package tech.ydb.spark.connector;
+package tech.ydb.spark.connector.impl;
 
 import java.util.Arrays;
 import java.util.HashMap;
@@ -10,6 +10,10 @@
 import org.apache.spark.sql.connector.catalog.TableChange;
 
 import tech.ydb.core.Status;
+import tech.ydb.spark.connector.YdbFieldInfo;
+import tech.ydb.spark.connector.YdbFieldType;
+import tech.ydb.spark.connector.YdbOptions;
+import tech.ydb.spark.connector.YdbTypes;
 import tech.ydb.table.Session;
 import tech.ydb.table.description.TableColumn;
 import tech.ydb.table.description.TableDescription;
@@ -22,7 +26,7 @@
  *
  * @author zinal
  */
-class YdbAlterTable extends YdbPropertyHelper {
+public class YdbAlterTable extends YdbPropertyHelper {
 
     final YdbTypes types;
     final String tablePath;
@@ -31,7 +35,7 @@ class YdbAlterTable extends YdbPropertyHelper {
     final Map<String, YdbFieldInfo> addColumns = new HashMap<>();
     final Set<String> removeColumns = new HashSet<>();
 
-    YdbAlterTable(YdbConnector connector, String tablePath) {
+    public YdbAlterTable(YdbConnector connector, String tablePath) {
         super(null);
         this.types = connector.getDefaultTypes();
         this.tablePath = tablePath;
@@ -43,7 +47,7 @@ class YdbAlterTable extends YdbPropertyHelper {
         }
     }
 
-    void prepare(TableChange.AddColumn change) {
+    public void prepare(TableChange.AddColumn change) {
         YdbFieldType yft = types.mapTypeSpark2Ydb(change.dataType());
         if (null == yft) {
             throw new UnsupportedOperationException("Unsupported data type for column: " + change.dataType());
@@ -68,7 +72,7 @@ void prepare(TableChange.AddColumn change) {
         }
     }
 
-    void prepare(TableChange.DeleteColumn change) {
+    public void prepare(TableChange.DeleteColumn change) {
         if (change.fieldNames().length != 1) {
             throw new UnsupportedOperationException("Illegal field name value: "
                     + Arrays.toString(change.fieldNames()));
@@ -88,7 +92,7 @@ void prepare(TableChange.DeleteColumn change) {
         }
     }
 
-    void prepare(TableChange.SetProperty change) {
+    public void prepare(TableChange.SetProperty change) {
         String property = change.property();
         if (!YdbOptions.TABLE_UPDATABLE.contains(property.toUpperCase())) {
             throw new UnsupportedOperationException("Unsupported property for table alteration: "
@@ -97,7 +101,7 @@ void prepare(TableChange.SetProperty change) {
         properties.put(property.toLowerCase(), change.value());
     }
 
-    void prepare(TableChange.RemoveProperty change) {
+    public void prepare(TableChange.RemoveProperty change) {
         String property = change.property();
         if (!YdbOptions.TABLE_UPDATABLE.contains(property.toUpperCase())) {
             throw new UnsupportedOperationException("Unsupported property for table alteration: "
@@ -142,7 +146,7 @@ private void applyProperty(String name, String value, PartitioningSettings ps) {
         }
     }
 
-    CompletableFuture<Status> run(Session session) {
+    public CompletableFuture<Status> run(Session session) {
         final AlterTableSettings settings = new AlterTableSettings();
         for (YdbFieldInfo yfi : addColumns.values()) {
             if (yfi.isNullable()) {

…h/ydb/spark/connector/YdbAlterTable.java …/spark/connector/impl/YdbAlterTable.javasrc/main/java/tech/ydb/spark/connector/YdbAlterTable.java renamed to src/main/java/tech/ydb/spark/connector/impl/YdbAlterTable.java src/main/java/tech/ydb/spark/connector/YdbAlterTable.java renamed to src/main/java/tech/ydb/spark/connector/impl/YdbAlterTable.java

@@ -1,4 +1,4 @@
-package tech.ydb.spark.connector;
+package tech.ydb.spark.connector.impl;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -13,6 +13,10 @@
 import tech.ydb.core.grpc.GrpcTransport;
 import tech.ydb.core.grpc.GrpcTransportBuilder;
 import tech.ydb.scheme.SchemeClient;
+import tech.ydb.spark.connector.YdbAuthMode;
+import tech.ydb.spark.connector.YdbIngestMethod;
+import tech.ydb.spark.connector.YdbOptions;
+import tech.ydb.spark.connector.YdbTypes;
 import tech.ydb.table.SessionRetryContext;
 import tech.ydb.table.TableClient;