@Override
  public void initialize(@Nullable Configuration configuration, Properties serDeProperties,
                         Properties partitionProperties) throws SerDeException {
    super.initialize(configuration, serDeProperties, partitionProperties);

    // HiveIcebergSerDe.initialize is called multiple places in Hive code:
    // - When we are trying to create a table - HiveDDL data is stored at the serDeProperties, but no Iceberg table
    // is created yet.
    // - When we are compiling the Hive query on HiveServer2 side - We only have table information (location/name),
    // and we have to read the schema using the table data. This is called multiple times so there is room for
    // optimizing here.
    // - When we are executing the Hive query in the execution engine - We do not want to load the table data on every
    // executor, but serDeProperties are populated by HiveIcebergStorageHandler.configureInputJobProperties() and
    // the resulting properties are serialized and distributed to the executors

    if (serDeProperties.get(InputFormatConfig.TABLE_SCHEMA) != null) {
      this.tableSchema = SchemaParser.fromJson(serDeProperties.getProperty(InputFormatConfig.TABLE_SCHEMA));
      if (serDeProperties.get(InputFormatConfig.PARTITION_SPEC) != null) {
        PartitionSpec spec =
            PartitionSpecParser.fromJson(tableSchema, serDeProperties.getProperty(InputFormatConfig.PARTITION_SPEC));
        this.partitionColumns = spec.fields().stream().map(PartitionField::name).collect(Collectors.toList());
      } else {
        this.partitionColumns = ImmutableList.of();
      }
    } else {
      try {
        Table table = IcebergTableUtil.getTable(configuration, serDeProperties);
        // always prefer the original table schema if there is one
        this.tableSchema = table.schema();
        this.partitionColumns = table.spec().fields().stream().map(PartitionField::name).collect(Collectors.toList());
        LOG.info("Using schema from existing table {}", SchemaParser.toJson(tableSchema));
      } catch (Exception e) {
        // During table creation we might not have the schema information from the Iceberg table, nor from the HMS
        // table. In this case we have to generate the schema using the serdeProperties which contains the info
        // provided in the CREATE TABLE query.

        if (serDeProperties.get("metadata_location") != null) {
          // If metadata location is provided, extract the schema details from it.
          try (FileIO fileIO = new HadoopFileIO(configuration)) {
            TableMetadata metadata = TableMetadataParser.read(fileIO, serDeProperties.getProperty("metadata_location"));
            this.tableSchema = metadata.schema();
            this.partitionColumns =
                metadata.spec().fields().stream().map(PartitionField::name).collect(Collectors.toList());
            // Validate no schema is provided via create command
            if (!getColumnNames().isEmpty() || !getPartitionColumnNames().isEmpty()) {
              throw new SerDeException("Column names can not be provided along with metadata location.");
            }
          }
        } else {
          boolean autoConversion = configuration.getBoolean(InputFormatConfig.SCHEMA_AUTO_CONVERSION, false);
          // If we can not load the table try the provided hive schema
          this.tableSchema = hiveSchemaOrThrow(e, autoConversion);
          // This is only for table creation, it is ok to have an empty partition column list
          this.partitionColumns = ImmutableList.of();
        }
        if (e instanceof NoSuchTableException &&
            HiveTableUtil.isCtas(serDeProperties) &&
            !Catalogs.hiveCatalog(configuration, serDeProperties)) {
          throw new SerDeException(CTAS_EXCEPTION_MSG);
        }
      }
    }

    this.projectedSchema =
        projectedSchema(configuration, serDeProperties.getProperty(Catalogs.NAME), tableSchema, jobConf);

    // Currently ClusteredWriter is used which requires that records are ordered by partition keys.
    // Here we ensure that SortedDynPartitionOptimizer will kick in and do the sorting.
    // TODO: remove once we have both Fanout and ClusteredWriter available: HIVE-25948
    HiveConf.setIntVar(configuration, HiveConf.ConfVars.HIVE_OPT_SORT_DYNAMIC_PARTITION_THRESHOLD, 1);
    HiveConf.setVar(configuration, HiveConf.ConfVars.DYNAMIC_PARTITIONING_MODE, "nonstrict");
    try {
      this.inspector = IcebergObjectInspector.create(projectedSchema);
    } catch (Exception e) {
      throw new SerDeException(e);
    }
  }
@Test
  public void testInitialize() throws IOException, SerDeException {
    File location = tmp.newFolder();
    Assert.assertTrue(location.delete());

    Configuration conf = new Configuration();

    Properties properties = new Properties();
    properties.setProperty("location", location.toString());
    properties.setProperty(InputFormatConfig.CATALOG_NAME, Catalogs.ICEBERG_HADOOP_TABLE_NAME);

    HadoopTables tables = new HadoopTables(conf);
    tables.create(schema, location.toString());

    HiveIcebergSerDe serDe = new HiveIcebergSerDe();
    serDe.initialize(conf, properties, new Properties());

    Assert.assertEquals(IcebergObjectInspector.create(schema), serDe.getObjectInspector());
  }