// Path: src/main/java/com/hortonworks/spark/sql/hive/llap/HiveWarehouseSession.java
// public interface HiveWarehouseSession {
// 
//     String HIVE_WAREHOUSE_CONNECTOR = "com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector";
//     String SPARK_DATASOURCES_PREFIX = "spark.datasource";
//     String HIVE_WAREHOUSE_POSTFIX = "hive.warehouse";
//     String CONF_PREFIX = SPARK_DATASOURCES_PREFIX + "." + HIVE_WAREHOUSE_POSTFIX;
// 
//     Dataset<Row> executeQuery(String sql);
//     Dataset<Row> q(String sql);
// 
//     Dataset<Row> execute(String sql);
// 
//     boolean executeUpdate(String sql);
// 
//     Dataset<Row> table(String sql);
// 
//     SparkSession session();
// 
//     void setDatabase(String name);
// 
//     Dataset<Row> showDatabases();
// 
//     Dataset<Row> showTables();
// 
//     Dataset<Row> describeTable(String table);
// 
//     void createDatabase(String database, boolean ifNotExists);
// 
//     CreateTableBuilder createTable(String tableName);
// 
//     void dropDatabase(String database, boolean ifExists, boolean cascade);
// 
//     void dropTable(String table, boolean ifExists, boolean purge);
// }


import java.util.Arrays;
import java.util.List;
import com.hortonworks.spark.sql.hive.llap.HiveWarehouseSession;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
import org.apache.spark.sql.sources.v2.DataSourceV2;
import org.apache.spark.sql.sources.v2.StreamWriteSupport;
import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
import org.apache.spark.sql.sources.v2.SessionConfigSupport;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
package com.hortonworks.spark.sql.hive.llap.streaming;



public class HiveStreamingDataSource implements DataSourceV2, StreamWriteSupport, SessionConfigSupport {
  private static Logger LOG = LoggerFactory.getLogger(HiveStreamingDataSource.class);

  @Override
  public StreamWriter createStreamWriter(final String queryId, final StructType schema, final OutputMode mode,
    final DataSourceOptions options) {
    return createDataSourceWriter(queryId, schema, options);
  }

  private HiveStreamingDataSourceWriter createDataSourceWriter(final String id, final StructType schema,
    final DataSourceOptions options) {
    String dbName = null;
    if(options.get("default.db").isPresent()) {
      dbName = options.get("default.db").get();
    } else {
      dbName = options.get("database").orElse("default");
    }
    String tableName = options.get("table").orElse(null);
    String partition = options.get("partition").orElse(null);
    List<String> partitionValues = partition == null ? null : Arrays.asList(partition.split(","));
    String metastoreUri = options.get("metastoreUri").orElse("thrift://localhost:9083");
    String metastoreKerberosPrincipal = options.get("metastoreKrbPrincipal").orElse(null);
    LOG.info("OPTIONS - database: {} table: {} partition: {} metastoreUri: {} metastoreKerberosPrincipal: {}",
      dbName, tableName, partition, metastoreUri, metastoreKerberosPrincipal);
    return new HiveStreamingDataSourceWriter(id, schema, dbName, tableName,
      partitionValues, metastoreUri, metastoreKerberosPrincipal);
  }

  @Override public String keyPrefix() {
// Path: src/main/java/com/hortonworks/spark/sql/hive/llap/HiveWarehouseSession.java
// public interface HiveWarehouseSession {
// 
//     String HIVE_WAREHOUSE_CONNECTOR = "com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector";
//     String SPARK_DATASOURCES_PREFIX = "spark.datasource";
//     String HIVE_WAREHOUSE_POSTFIX = "hive.warehouse";
//     String CONF_PREFIX = SPARK_DATASOURCES_PREFIX + "." + HIVE_WAREHOUSE_POSTFIX;
// 
//     Dataset<Row> executeQuery(String sql);
//     Dataset<Row> q(String sql);
// 
//     Dataset<Row> execute(String sql);
// 
//     boolean executeUpdate(String sql);
// 
//     Dataset<Row> table(String sql);
// 
//     SparkSession session();
// 
//     void setDatabase(String name);
// 
//     Dataset<Row> showDatabases();
// 
//     Dataset<Row> showTables();
// 
//     Dataset<Row> describeTable(String table);
// 
//     void createDatabase(String database, boolean ifNotExists);
// 
//     CreateTableBuilder createTable(String tableName);
// 
//     void dropDatabase(String database, boolean ifExists, boolean cascade);
// 
//     void dropTable(String table, boolean ifExists, boolean purge);
// }


// Path: src/main/java/com/hortonworks/spark/sql/hive/llap/streaming/HiveStreamingDataSource.java
import java.util.Arrays;
import java.util.List;
import com.hortonworks.spark.sql.hive.llap.HiveWarehouseSession;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
import org.apache.spark.sql.sources.v2.DataSourceV2;
import org.apache.spark.sql.sources.v2.StreamWriteSupport;
import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
import org.apache.spark.sql.sources.v2.SessionConfigSupport;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package com.hortonworks.spark.sql.hive.llap.streaming;



public class HiveStreamingDataSource implements DataSourceV2, StreamWriteSupport, SessionConfigSupport {
  private static Logger LOG = LoggerFactory.getLogger(HiveStreamingDataSource.class);

  @Override
  public StreamWriter createStreamWriter(final String queryId, final StructType schema, final OutputMode mode,
    final DataSourceOptions options) {
    return createDataSourceWriter(queryId, schema, options);
  }

  private HiveStreamingDataSourceWriter createDataSourceWriter(final String id, final StructType schema,
    final DataSourceOptions options) {
    String dbName = null;
    if(options.get("default.db").isPresent()) {
      dbName = options.get("default.db").get();
    } else {
      dbName = options.get("database").orElse("default");
    }
    String tableName = options.get("table").orElse(null);
    String partition = options.get("partition").orElse(null);
    List<String> partitionValues = partition == null ? null : Arrays.asList(partition.split(","));
    String metastoreUri = options.get("metastoreUri").orElse("thrift://localhost:9083");
    String metastoreKerberosPrincipal = options.get("metastoreKrbPrincipal").orElse(null);
    LOG.info("OPTIONS - database: {} table: {} partition: {} metastoreUri: {} metastoreKerberosPrincipal: {}",
      dbName, tableName, partition, metastoreUri, metastoreKerberosPrincipal);
    return new HiveStreamingDataSourceWriter(id, schema, dbName, tableName,
      partitionValues, metastoreUri, metastoreKerberosPrincipal);
  }

  @Override public String keyPrefix() {
    return HiveWarehouseSession.HIVE_WAREHOUSE_POSTFIX;