Merge branch 'main' of github.com:hypertrace/document-store into bugfix_setSubDocOp

Author: suddendust

document-store/src/integrationTest/java/org/hypertrace/core/documentstore/MongoFlatPgConsistencyTest.java

@@ -94,7 +94,8 @@ public DatastoreConfig convert(final Config config) {
               connectionConfig.applicationName(),
               connectionConfig.connectionPoolConfig(),
               connectionConfig.queryTimeout(),
-              connectionConfig.customParameters()) {
+              connectionConfig.customParameters(),
+              connectionConfig.collectionConfigs()) {
             @Override
             public String toConnectionString() {
               return config.hasPath("url")

document-store/src/main/java/org/hypertrace/core/documentstore/TypesafeDatastoreConfigAdapter.java

@@ -75,12 +75,21 @@ public static class ConnectionConfigBuilder {
     String applicationName = DEFAULT_APP_NAME;
     String replicaSet;
     Map<String, String> customParameters = new HashMap<>();
+    Map<String, org.hypertrace.core.documentstore.model.config.postgres.CollectionConfig>
+        collectionConfigs = new HashMap<>();
 
     public ConnectionConfigBuilder customParameter(String key, String value) {
       this.customParameters.put(key, value);
       return this;
     }
 
+    public ConnectionConfigBuilder collectionConfig(
+        String collectionName,
+        org.hypertrace.core.documentstore.model.config.postgres.CollectionConfig config) {
+      this.collectionConfigs.put(collectionName, config);
+      return this;
+    }
+
     ConnectionPoolConfig connectionPoolConfig;
     AggregatePipelineMode aggregationPipelineMode = AggregatePipelineMode.DEFAULT_ALWAYS;
     DataFreshness dataFreshness = DataFreshness.SYSTEM_DEFAULT;
@@ -127,7 +136,8 @@ public ConnectionConfig build() {
               applicationName,
               connectionPoolConfig,
               queryTimeout,
-              customParameters);
+              customParameters,
+              collectionConfigs);
       }
 
       throw new IllegalArgumentException("Unsupported database type: " + type);

document-store/src/main/java/org/hypertrace/core/documentstore/model/config/ConnectionConfig.java

@@ -10,12 +10,14 @@
 import org.hypertrace.core.documentstore.model.config.ConnectionPoolConfig.ConnectionPoolConfigBuilder;
 import org.hypertrace.core.documentstore.model.config.DatastoreConfig.DatastoreConfigBuilder;
 import org.hypertrace.core.documentstore.model.config.Endpoint.EndpointBuilder;
+import org.hypertrace.core.documentstore.model.config.postgres.TimestampFieldsConfig;
 import org.hypertrace.core.documentstore.model.options.DataFreshness;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 @Value
 public class TypesafeConfigDatastoreConfigExtractor {
+
   private static final Logger LOGGER =
       LoggerFactory.getLogger(TypesafeConfigDatastoreConfigExtractor.class);
   private static final String DEFAULT_HOST_KEY = "host";
@@ -35,6 +37,10 @@ public class TypesafeConfigDatastoreConfigExtractor {
   private static final String DEFAULT_QUERY_TIMEOUT_KEY = "queryTimeout";
   private static final String DEFAULT_CONNECTION_TIMEOUT_KEY = "connectionTimeout";
   private static final String DEFAULT_CUSTOM_PARAMETERS_PREFIX = "customParams";
+  private static final String DEFAULT_COLLECTION_CONFIGS_PREFIX = "collectionConfigs";
+  private static final String TIMESTAMP_FIELDS_KEY = "timestampFields";
+  private static final String TIMESTAMP_CREATED_KEY = "created";
+  private static final String TIMESTAMP_LAST_UPDATED_KEY = "lastUpdated";
 
   @NonNull Config config;
   DatastoreConfigBuilder datastoreConfigBuilder;
@@ -82,7 +88,8 @@ private TypesafeConfigDatastoreConfigExtractor(
         .dataFreshnessKey(DEFAULT_DATA_FRESHNESS_KEY)
         .queryTimeoutKey(DEFAULT_QUERY_TIMEOUT_KEY)
         .connectionTimeoutKey(DEFAULT_CONNECTION_TIMEOUT_KEY)
-        .customParametersKey(DEFAULT_CUSTOM_PARAMETERS_PREFIX);
+        .customParametersKey(DEFAULT_CUSTOM_PARAMETERS_PREFIX)
+        .collectionConfigsKey(dataStoreType + "." + DEFAULT_COLLECTION_CONFIGS_PREFIX);
   }
 
   public static TypesafeConfigDatastoreConfigExtractor from(
@@ -253,6 +260,53 @@ public TypesafeConfigDatastoreConfigExtractor connectionTimeoutKey(@NonNull fina
     return this;
   }
 
+  public TypesafeConfigDatastoreConfigExtractor collectionConfigsKey(@NonNull final String key) {
+    if (!config.hasPath(key)) {
+      return this;
+    }
+
+    try {
+      Config collectionsConfig = config.getConfig(key);
+      collectionsConfig
+          .root()
+          .keySet()
+          .forEach(
+              collectionName -> {
+                Config collectionConfig = collectionsConfig.getConfig(collectionName);
+                org.hypertrace.core.documentstore.model.config.postgres.CollectionConfig
+                        .CollectionConfigBuilder
+                    builder =
+                        org.hypertrace.core.documentstore.model.config.postgres.CollectionConfig
+                            .builder();
+
+                if (collectionConfig.hasPath(TIMESTAMP_FIELDS_KEY)) {
+                  builder.timestampFields(
+                      parseTimestampFieldsConfig(collectionConfig.getConfig(TIMESTAMP_FIELDS_KEY)));
+                }
+
+                connectionConfigBuilder.collectionConfig(collectionName, builder.build());
+              });
+    } catch (Exception e) {
+      LOGGER.warn(
+          "Collection configs key '{}' exists but could not be parsed: {}", key, e.getMessage());
+    }
+
+    return this;
+  }
+
+  private TimestampFieldsConfig parseTimestampFieldsConfig(Config timestampConfig) {
+    TimestampFieldsConfig.TimestampFieldsConfigBuilder tsBuilder = TimestampFieldsConfig.builder();
+
+    if (timestampConfig.hasPath(TIMESTAMP_CREATED_KEY)) {
+      tsBuilder.created(timestampConfig.getString(TIMESTAMP_CREATED_KEY));
+    }
+    if (timestampConfig.hasPath(TIMESTAMP_LAST_UPDATED_KEY)) {
+      tsBuilder.lastUpdated(timestampConfig.getString(TIMESTAMP_LAST_UPDATED_KEY));
+    }
+
+    return tsBuilder.build();
+  }
+
   public DatastoreConfig extract() {
     if (connectionConfigBuilder.endpoints().isEmpty()
         && !Endpoint.builder().build().equals(endpointBuilder.build())) {
@@ -265,6 +319,7 @@ public DatastoreConfig extract() {
                 .connectionPoolConfig(connectionPoolConfigBuilder.build())
                 .credentials(connectionCredentialsBuilder.build())
                 .customParameters(connectionConfigBuilder.customParameters())
+                .collectionConfigs(connectionConfigBuilder.collectionConfigs())
                 .build())
         .build();
   }

document-store/src/main/java/org/hypertrace/core/documentstore/model/config/TypesafeConfigDatastoreConfigExtractor.java

@@ -0,0 +1,22 @@
+package org.hypertrace.core.documentstore.model.config.postgres;
+
+import java.util.Optional;
+import javax.annotation.Nullable;
+import lombok.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+public class CollectionConfig {
+
+  /** Timestamp field configuration for auto-managed created/updated timestamps */
+  @Nullable TimestampFieldsConfig timestampFields;
+
+  public Optional<TimestampFieldsConfig> getTimestampFields() {
+    return Optional.ofNullable(timestampFields);
+  }
+
+  public static CollectionConfig empty() {
+    return CollectionConfig.builder().build();
+  }
+}

document-store/src/main/java/org/hypertrace/core/documentstore/model/config/postgres/CollectionConfig.java

@@ -5,6 +5,7 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -46,6 +47,7 @@ public class PostgresConnectionConfig extends ConnectionConfig {
   @NonNull Duration queryTimeout;
   @NonNull Duration schemaCacheExpiry;
   @NonNull Duration schemaRefreshCooldown;
+  @NonNull Map<String, CollectionConfig> collectionConfigs;
 
   public static ConnectionConfigBuilder builder() {
     return ConnectionConfig.builder().type(DatabaseType.POSTGRES);
@@ -58,7 +60,8 @@ public PostgresConnectionConfig(
       @NonNull final String applicationName,
       @Nullable final ConnectionPoolConfig connectionPoolConfig,
       @NonNull final Duration queryTimeout,
-      @NonNull final Map<String, String> customParameters) {
+      @NonNull final Map<String, String> customParameters,
+      @Nullable final Map<String, CollectionConfig> collectionConfigs) {
     super(
         ensureSingleEndpoint(endpoints),
         getDatabaseOrDefault(database),
@@ -69,6 +72,7 @@ public PostgresConnectionConfig(
     this.queryTimeout = queryTimeout;
     this.schemaCacheExpiry = extractSchemaCacheExpiry(customParameters);
     this.schemaRefreshCooldown = extractSchemaRefreshCooldown(customParameters);
+    this.collectionConfigs = collectionConfigs != null ? collectionConfigs : Collections.emptyMap();
   }
 
   public String toConnectionString() {
@@ -157,4 +161,8 @@ private static Duration extractSchemaRefreshCooldown(final Map<String, String> c
         .map(Duration::ofMillis)
         .orElse(DEFAULT_SCHEMA_REFRESH_COOLDOWN);
   }
+
+  public Optional<CollectionConfig> getCollectionConfig(String collectionName) {
+    return Optional.ofNullable(collectionConfigs.get(collectionName));
+  }
 }

document-store/src/main/java/org/hypertrace/core/documentstore/model/config/postgres/PostgresConnectionConfig.java

@@ -0,0 +1,35 @@
+package org.hypertrace.core.documentstore.model.config.postgres;
+
+import java.util.Optional;
+import javax.annotation.Nullable;
+import lombok.Builder;
+import lombok.Value;
+
+/**
+ * Configuration for auto-managed timestamp fields in a Postgres collection. Specifies which columns
+ * should be automatically populated with creation and last update timestamps.
+ */
+@Value
+@Builder
+public class TimestampFieldsConfig {
+
+  /**
+   * Column name for the creation timestamp. This field is set once when a document is created and
+   * preserved on updates.
+   */
+  @Nullable String created;
+
+  /**
+   * Column name for the last update timestamp. This field is updated every time a document is
+   * created or modified.
+   */
+  @Nullable String lastUpdated;
+
+  public Optional<String> getCreated() {
+    return Optional.ofNullable(created);
+  }
+
+  public Optional<String> getLastUpdated() {
+    return Optional.ofNullable(lastUpdated);
+  }
+}