public static void applyTimestampIndex(TableConfig tableConfig, Schema schema) {
    if (tableConfig.getFieldConfigList() == null) {
      return;
    }

    Map<String, List<TimestampIndexGranularity>> timestampIndexConfigs = new HashMap<>();
    for (FieldConfig fieldConfig : tableConfig.getFieldConfigList()) {
      TimestampConfig timestampConfig = fieldConfig.getTimestampConfig();
      if (timestampConfig == null || CollectionUtils.isEmpty(timestampConfig.getGranularities())) {
        continue;
      }
      timestampIndexConfigs.put(fieldConfig.getName(), timestampConfig.getGranularities());
    }
    if (timestampIndexConfigs.isEmpty()) {
      return;
    }

    // Synchronize on table config object to prevent concurrent modification
    //noinspection SynchronizationOnLocalVariableOrMethodParameter
    synchronized (tableConfig) {
      // Check if the updates are already applied
      boolean schemaApplied;
      boolean tableConfigApplied;
      Map.Entry<String, List<TimestampIndexGranularity>> sampleEntry =
          timestampIndexConfigs.entrySet().iterator().next();
      String sampleTimestampColumnWithGranularity =
          getColumnWithGranularity(sampleEntry.getKey(), sampleEntry.getValue().get(0));
      schemaApplied = schema.hasColumn(sampleTimestampColumnWithGranularity);
      IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
      List<String> rangeIndexColumns = indexingConfig.getRangeIndexColumns();
      tableConfigApplied =
          rangeIndexColumns != null && rangeIndexColumns.contains(sampleTimestampColumnWithGranularity);
      if (schemaApplied && tableConfigApplied) {
        return;
      }

      // Apply TIMESTAMP index
      List<TransformConfig> transformConfigs = null;
      if (!tableConfigApplied) {
        if (rangeIndexColumns == null) {
          rangeIndexColumns = new ArrayList<>();
          indexingConfig.setRangeIndexColumns(rangeIndexColumns);
        }
        IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
        if (ingestionConfig == null) {
          ingestionConfig = new IngestionConfig();
          tableConfig.setIngestionConfig(ingestionConfig);
        }
        transformConfigs = ingestionConfig.getTransformConfigs();
        if (transformConfigs == null) {
          transformConfigs = new ArrayList<>();
          ingestionConfig.setTransformConfigs(transformConfigs);
        }
      }
      for (Map.Entry<String, List<TimestampIndexGranularity>> entry : timestampIndexConfigs.entrySet()) {
        String timestampColumn = entry.getKey();
        for (TimestampIndexGranularity granularity : entry.getValue()) {
          String columnWithGranularity = getColumnWithGranularity(timestampColumn, granularity);
          if (!schemaApplied) {
            schema.addField(getFieldSpecWithGranularity(columnWithGranularity, granularity));
          }
          if (!tableConfigApplied) {
            transformConfigs.add(
                new TransformConfig(columnWithGranularity, getTransformExpression(timestampColumn, granularity)));
            rangeIndexColumns.add(columnWithGranularity);
          }
        }
      }
    }
  }
@Test
  public void testApplyTimestampIndex() {
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setFieldConfigList(
        Arrays.asList(
            new FieldConfig("ts1", FieldConfig.EncodingType.DICTIONARY, FieldConfig.IndexType.TIMESTAMP, null, null,
                new TimestampConfig(Arrays.asList(TimestampIndexGranularity.SECOND, TimestampIndexGranularity.MINUTE,
                    TimestampIndexGranularity.HOUR)), null),
            new FieldConfig("ts2", FieldConfig.EncodingType.RAW, FieldConfig.IndexType.TIMESTAMP, null, null,
                new TimestampConfig(Arrays.asList(TimestampIndexGranularity.DAY, TimestampIndexGranularity.WEEK,
                    TimestampIndexGranularity.MONTH)), null))).build();
    Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable")
        .addDateTime("ts1", DataType.TIMESTAMP, "TIMESTAMP", "1:MILLISECONDS")
        .addDateTime("ts2", DataType.TIMESTAMP, "TIMESTAMP", "1:MILLISECONDS").build();

    // Apply TIMESTAMP index multiple times should get the same result
    for (int i = 0; i < 5; i++) {
      TimestampIndexUtils.applyTimestampIndex(tableConfig, schema);

      // Check schema
      assertEquals(schema.size(), 8);
      FieldSpec ts1SecondFieldSpec = schema.getFieldSpecFor("$ts1$SECOND");
      assertNotNull(ts1SecondFieldSpec);
      assertTrue(ts1SecondFieldSpec instanceof DateTimeFieldSpec);
      DateTimeFieldSpec ts1SecondDateTimeFieldSpec = (DateTimeFieldSpec) ts1SecondFieldSpec;
      assertEquals(ts1SecondDateTimeFieldSpec.getDataType(), DataType.TIMESTAMP);
      assertEquals(ts1SecondDateTimeFieldSpec.getFormat(), "TIMESTAMP");
      assertEquals(ts1SecondDateTimeFieldSpec.getGranularity(), "1:SECONDS");

      // Check ingestion transform
      assertNotNull(tableConfig.getIngestionConfig());
      List<TransformConfig> transformConfigs = tableConfig.getIngestionConfig().getTransformConfigs();
      assertNotNull(transformConfigs);
      assertEquals(transformConfigs.size(), 6);
      Set<String> transformColumns = new HashSet<>();
      for (TransformConfig transformConfig : transformConfigs) {
        String columnName = transformConfig.getColumnName();
        assertTrue(transformColumns.add(columnName));
        if (columnName.equals("$ts2$DAY")) {
          assertEquals(transformConfig.getTransformFunction(), "dateTrunc('DAY',\"ts2\")");
        }
      }
      assertEquals(transformColumns, new HashSet<>(
          Arrays.asList("$ts1$SECOND", "$ts1$MINUTE", "$ts1$HOUR", "$ts2$DAY", "$ts2$WEEK", "$ts2$MONTH")));

      // Check range index
      List<String> rangeIndexColumns = tableConfig.getIndexingConfig().getRangeIndexColumns();
      assertNotNull(rangeIndexColumns);
      assertEquals(new HashSet<>(rangeIndexColumns), transformColumns);
    }
  }