Fixed inconsistend reads

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

@@ -13,7 +13,6 @@
 import java.util.Objects;
 
 import com.google.common.io.ByteStreams;
-import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,7 +34,7 @@
  *
  * @author Aleksandr Gorshenin
  */
-public class YdbContext implements Serializable {
+public class YdbContext implements Serializable, AutoCloseable {
     private static final Logger logger = LoggerFactory.getLogger(YdbContext.class);
 
     private static final long serialVersionUID = 6522842483896983993L;
@@ -62,7 +61,7 @@ public class YdbContext implements Serializable {
 
     private final int sessionPoolSize;
 
-    public YdbContext(CaseInsensitiveStringMap options) {
+    public YdbContext(Map<String, String> options) {
         this.connectionString = ConnectionOption.URL.read(options);
         if (connectionString  == null || this.connectionString .trim().isEmpty()) {
             throw new IllegalArgumentException("Incorrect value for property " + ConnectionOption.URL);
@@ -121,6 +120,11 @@ public boolean equals(Object other) {
                 sessionPoolSize == sessionPoolSize;
     }
 
+    @Override
+    public void close() {
+        YdbRegistry.closeExecutor(this);
+    }
+
     public YdbExecutor getExecutor() {
         return YdbRegistry.getOrCreate(this, YdbContext::createExecutor);
     }

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

@@ -32,4 +32,11 @@ public static void closeAll() {
             }
         }
     }
+
+    public static void closeExecutor(YdbContext ctx) {
+        YdbExecutor executor = EXECUTORS.remove(ctx);
+        if (executor != null) {
+            executor.close();
+        }
+    }
 }

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

@@ -21,6 +21,8 @@
  * @author zinal
  */
 public class YdbTableProvider implements TableProvider, DataSourceRegister {
+    private static final String SPARK_PATH_OPTION = "path";
+
     @Override
     public String shortName() {
         return "ydb";
@@ -34,10 +36,16 @@ public boolean supportsExternalMetadata() {
     private String exractTableName(CaseInsensitiveStringMap options) {
         // Check that table path is provided
         String table = OperationOption.DBTABLE.read(options);
-        if (table == null || table.trim().length() == 0) {
-            throw new IllegalArgumentException("Missing property: " + OperationOption.DBTABLE);
+        if (table != null && !table.trim().isEmpty()) {
+            return table.trim();
+        }
+
+        String path = options.get(SPARK_PATH_OPTION);
+        if (path != null && !path.trim().isEmpty()) {
+            return path.trim();
         }
-        return table.trim();
+
+        throw new IllegalArgumentException("Missing property: " + OperationOption.DBTABLE);
     }
 
     @Override