@Override
  public void initialize(@Nullable Configuration configuration, Properties serDeProperties)
      throws SerDeException {
    // 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((String) serDeProperties.get(InputFormatConfig.TABLE_SCHEMA));
    } else {
      try {
        // always prefer the original table schema if there is one
        this.tableSchema = Catalogs.loadTable(configuration, serDeProperties).schema();
        LOG.info("Using schema from existing table {}", SchemaParser.toJson(tableSchema));
      } catch (Exception e) {
        boolean autoConversion =
            configuration.getBoolean(InputFormatConfig.SCHEMA_AUTO_CONVERSION, false);
        // If we can not load the table try the provided hive schema
        this.tableSchema = hiveSchemaOrThrow(serDeProperties, e, autoConversion);
      }
    }

    Schema projectedSchema;
    if (serDeProperties.get(HiveIcebergStorageHandler.WRITE_KEY) != null) {
      // when writing out data, we should not do projection pushdown
      projectedSchema = tableSchema;
    } else {
      configuration.setBoolean(InputFormatConfig.CASE_SENSITIVE, false);
      String[] selectedColumns = ColumnProjectionUtils.getReadColumnNames(configuration);
      // When same table is joined multiple times, it is possible some selected columns are
      // duplicated,
      // in this case wrong recordStructField position leads wrong value or
      // ArrayIndexOutOfBoundException
      String[] distinctSelectedColumns =
          Arrays.stream(selectedColumns).distinct().toArray(String[]::new);
      projectedSchema =
          distinctSelectedColumns.length > 0
              ? tableSchema.caseInsensitiveSelect(distinctSelectedColumns)
              : tableSchema;
      // the input split mapper handles does not belong to this table
      // it is necessary to ensure projectedSchema equals to tableSchema,
      // or we cannot find selectOperator's column from inspector
      if (projectedSchema.columns().size() != distinctSelectedColumns.length) {
        projectedSchema = tableSchema;
      }
    }

    try {
      this.inspector = IcebergObjectInspector.create(projectedSchema);
    } catch (Exception e) {
      throw new SerDeException(e);
    }
  }
@Test
  public void testInitialize() throws IOException, SerDeException {
    File location = tmp.toFile();
    assertThat(location.delete()).isTrue();

    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);

    assertThat(serDe.getObjectInspector()).isEqualTo(IcebergObjectInspector.create(SCHEMA));
  }