Support for replacing varchar and varbinary types in external tables (#…

…266)

* Add nameservice to hadoopimpersonationconfig

* Remove condition that restricts column replacement to only UNKNOWN

* revert ha change in this branch

connector/src/main/scala/com/vertica/spark/datasource/core/VerticaDistributedFilesystemWritePipe.scala

@@ -240,14 +240,22 @@ class VerticaDistributedFilesystemWritePipe(val config: DistributedFilesystemWri
  }
 
  logger.debug("The infer statement is: " + inferStatement)
-
  jdbcLayer.query(inferStatement) match {
  case Left(err) => Left(InferExternalTableSchemaError(err))
  case Right(resultSet) =>
  try {
  val iterate = resultSet.next
  val createExternalTableStatement = resultSet.getString("INFER_EXTERNAL_TABLE_DDL")
 
+ val containsVarchar = createExternalTableStatement.toLowerCase.contains("varchar")
+ val containsVarbinary = createExternalTableStatement.toLowerCase.contains("varbinary")
+
+ if(containsVarchar || containsVarbinary) {
+ logger.warn("The parquet data contains a column of type varchar or varbinary. " +
+ "Lengths of these column types cannot be determined from the data and will truncate to the default length (80). " +
+ "Please provide a partial schema with StringType to replace varchar and BinaryType to replace varbinary, or manually create an external table.")
+ }
+
  if(inferStatement.contains(EscapeUtils.sqlEscape(s"${config.fileStoreConfig.externalTableAddress.stripSuffix("/")}/**/*.parquet"))) {
  logger.info("Inferring partial schema from dataframe")
  schemaTools.inferExternalTableSchema(createExternalTableStatement, config.schema, tableName)

connector/src/main/scala/com/vertica/spark/util/schema/SchemaTools.scala

@@ -444,22 +444,20 @@ class SchemaTools extends SchemaToolsInterface {
  val schemaList = schemaString.split(",").toList
 
  val updatedSchema: String = schemaList.map(col => {
- if(col.contains(unknown)) {
- val indexOfFirstDoubleQuote = col.indexOf("\"")
- val indexOfSpace = col.indexOf(" ", indexOfFirstDoubleQuote)
- val colName = col.substring(indexOfFirstDoubleQuote, indexOfSpace)
+ val indexOfFirstDoubleQuote = col.indexOf("\"")
+ val indexOfSpace = col.indexOf(" ", indexOfFirstDoubleQuote)
+ val colName = col.substring(indexOfFirstDoubleQuote, indexOfSpace)
 
-  val fieldType = schema.collect {
-  case field if(addDoubleQuotes(field.name) == colName) => field.dataType.simpleString
-  }
-  if(fieldType.nonEmpty) {
-  colName + " " + fieldType.head
-  }
- else {
-  col
- }
+ val fieldType = schema.collect {
+ case field if(addDoubleQuotes(field.name) == colName) => field.dataType.simpleString
+ }
+ if(fieldType.nonEmpty) {
+ colName + " " + fieldType.head
+ }
+
+ else {
+ col
  }
- else { col }
  }).mkString(",")
 
  if(updatedSchema.contains(unknown)) {

connector/src/test/scala/com/vertica/spark/util/schema/SchemaToolsTest.scala

@@ -543,7 +543,7 @@ class SchemaToolsTests extends AnyFlatSpec with BeforeAndAfterAll with MockFacto
  val createExternalTableStmt = "create external table \"sales\"(" +
  "\"tx_id\" int," +
  "\"date\" UNKNOWN," +
- "\"region\" UNKNOWN" +
+ "\"region\" varchar" +
  ") as copy from \'/data/\' parquet"
  val schemaTools = new SchemaTools
  schemaTools.inferExternalTableSchema(createExternalTableStmt, schema, "sales") match {