WIP

suddendust · suddendust · commit b6c0237b88c7 · 2026-04-12T22:08:23.000+05:30
diff --git a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java
@@ -228,255 +228,8 @@ public boolean updateSubDoc(Key key, String subDocPath, Document subDocument) {
   }
 
   @Override
-  public BulkUpdateResult bulkUpdateSubDocs(Map<Key, Map<String, Document>> documents)
-      throws Exception {
-    if (documents == null || documents.isEmpty()) {
-      return new BulkUpdateResult(0);
-    }
-
-    String tableName = tableIdentifier.getTableName();
-    String pkColumn = getPKForTable(tableName);
-    String quotedPkColumn = PostgresUtils.wrapFieldNamesWithDoubleQuotes(pkColumn);
-
-    Set<Key> updatedKeys = new HashSet<>();
-
-    try (Connection connection = client.getTransactionalConnection()) {
-      try {
-        // Lock all rows upfront with SELECT ... FOR UPDATE to ensure atomicity
-        String lockSql =
-            String.format(
-                "SELECT %s FROM %s WHERE %s = ANY(?) FOR UPDATE",
-                quotedPkColumn, tableIdentifier, quotedPkColumn);
-
-        String[] keyArray = documents.keySet().stream().map(Key::toString).toArray(String[]::new);
-        Array sqlArray = connection.createArrayOf("text", keyArray);
-
-        try (PreparedStatement lockPs = connection.prepareStatement(lockSql)) {
-          lockPs.setArray(1, sqlArray);
-          lockPs.executeQuery(); // Acquire locks on all rows
-        }
-
-        // Now execute updates for each key
-        for (Map.Entry<Key, Map<String, Document>> entry : documents.entrySet()) {
-          Key key = entry.getKey();
-          Map<String, Document> subDocUpdates = entry.getValue();
-
-          if (subDocUpdates == null || subDocUpdates.isEmpty()) {
-            continue;
-          }
-
-          boolean updated =
-              executeSubDocUpdatesForKey(connection, key, subDocUpdates, tableName, quotedPkColumn);
-          if (updated) {
-            updatedKeys.add(key);
-          }
-        }
-
-        // Update lastUpdatedTime within the same transaction
-        updateLastUpdatedTimeForKeys(connection, updatedKeys, tableName, quotedPkColumn);
-
-        connection.commit();
-      } catch (Exception e) {
-        connection.rollback();
-        throw e;
-      }
-    } catch (SQLException e) {
-      LOGGER.error("SQLException in bulkUpdateSubDocs", e);
-      throw e;
-    }
-
-    return new BulkUpdateResult(updatedKeys.size());
-  }
-
-  /**
-   * Executes sub-document updates for a single key.
-   *
-   * @return true if at least one update was successful
-   */
-  private boolean executeSubDocUpdatesForKey(
-      Connection connection,
-      Key key,
-      Map<String, Document> subDocUpdates,
-      String tableName,
-      String quotedPkColumn)
-      throws SQLException {
-
-    // Group updates by resolved column
-    Map<String, List<SubDocUpdateEntry>> updatesByColumn = new LinkedHashMap<>();
-
-    for (Map.Entry<String, Document> subDocEntry : subDocUpdates.entrySet()) {
-      String subDocPath = subDocEntry.getKey();
-      Document subDocument = subDocEntry.getValue();
-
-      Optional<String> columnName = resolveColumnName(subDocPath, tableName);
-      if (columnName.isEmpty()) {
-        if (missingColumnStrategy == MissingColumnStrategy.THROW) {
-          throw new IllegalArgumentException(
-              "Column not found in schema for path: "
-                  + subDocPath
-                  + " and missing column strategy is configured to: "
-                  + missingColumnStrategy);
-        }
-        LOGGER.warn("Skipping update for unresolved path: {}", subDocPath);
-        continue;
-      }
-
-      updatesByColumn
-          .computeIfAbsent(columnName.get(), k -> new ArrayList<>())
-          .add(new SubDocUpdateEntry(subDocPath, subDocument));
-    }
-
-    if (updatesByColumn.isEmpty()) {
-      return false;
-    }
-
-    // Build SET clause fragments - one per column
-    List<String> setFragments = new ArrayList<>();
-    List<Object> params = new ArrayList<>();
-
-    for (Map.Entry<String, List<SubDocUpdateEntry>> columnEntry : updatesByColumn.entrySet()) {
-      String columnName = columnEntry.getKey();
-      List<SubDocUpdateEntry> columnUpdates = columnEntry.getValue();
-
-      PostgresColumnMetadata colMeta =
-          schemaRegistry.getColumnOrRefresh(tableName, columnName).orElseThrow();
-
-      // Track the current expression for chaining nested JSONB updates
-      String currentExpr = String.format("\"%s\"", columnName);
-      String fragment = null;
-
-      for (SubDocUpdateEntry updateEntry : columnUpdates) {
-        String fullPath = updateEntry.path;
-        String[] nestedPath = getNestedPath(fullPath, columnName);
-        boolean isTopLevel = nestedPath.length == 0;
-
-        if (isTopLevel) {
-          // Top-level column update - set the entire column value
-          PostgresDataType colType = colMeta.getPostgresType();
-          if (colType == PostgresDataType.JSONB) {
-            // JSONB columns need explicit cast
-            fragment = String.format("\"%s\" = ?::jsonb", columnName);
-            params.add(updateEntry.document.toJson());
-          } else {
-            fragment = String.format("\"%s\" = ?", columnName);
-            params.add(prepareValueForColumn(updateEntry.document, colMeta));
-          }
-        } else {
-          // Nested JSONB path update using jsonb_set
-          String jsonPath = formatJsonbPath(nestedPath);
-          String jsonValue = updateEntry.document.toJson();
-          String valueExpr =
-              String.format("jsonb_set(%s, '%s'::text[], ?::jsonb)", currentExpr, jsonPath);
-          currentExpr = valueExpr;
-          fragment = String.format("\"%s\" = %s", columnName, valueExpr);
-          params.add(jsonValue);
-        }
-      }
-
-      if (fragment != null) {
-        setFragments.add(fragment);
-      }
-    }
-
-    if (setFragments.isEmpty()) {
-      return false;
-    }
-
-    // Build and execute UPDATE SQL
-    String sql =
-        String.format(
-            "UPDATE %s SET %s WHERE %s = ?",
-            tableIdentifier, String.join(", ", setFragments), quotedPkColumn);
-
-    params.add(key.toString());
-
-    LOGGER.debug("Executing sub-doc update SQL: {}", sql);
-
-    try (PreparedStatement ps = connection.prepareStatement(sql)) {
-      int idx = 1;
-      for (Object param : params) {
-        ps.setObject(idx++, param);
-      }
-      int rowsUpdated = ps.executeUpdate();
-      return rowsUpdated > 0;
-    }
-  }
-
-  /**
-   * Prepares a Document value for insertion into a column based on its type. For JSONB columns,
-   * returns the JSON string. For other types, extracts the appropriate scalar value.
-   */
-  private Object prepareValueForColumn(Document document, PostgresColumnMetadata colMeta)
-      throws SQLException {
-    try {
-      JsonNode jsonNode = MAPPER.readTree(document.toJson());
-      PostgresDataType type = colMeta.getPostgresType();
-
-      if (type == PostgresDataType.JSONB) {
-        return document.toJson();
-      }
-
-      // For non-JSONB columns, extract the scalar value
-      return extractValue(jsonNode, type, colMeta.isArray());
-    } catch (IOException e) {
-      throw new SQLException("Failed to parse document for column: " + colMeta.getName(), e);
-    }
-  }
-
-  /** Formats a nested path array into PostgreSQL jsonb path format: {path1,path2,path3} */
-  private String formatJsonbPath(String[] nestedPath) {
-    return "{" + String.join(",", nestedPath) + "}";
-  }
-
-  /**
-   * Updates the lastUpdatedTime column for the given keys using the provided connection. This
-   * allows the update to be part of an existing transaction.
-   */
-  private void updateLastUpdatedTimeForKeys(
-      Connection connection, Set<Key> keys, String tableName, String quotedPkColumn)
-      throws SQLException {
-    if (keys.isEmpty() || lastUpdatedTsColumn == null) {
-      return;
-    }
-
-    String quotedTsColumn = PostgresUtils.wrapFieldNamesWithDoubleQuotes(lastUpdatedTsColumn);
-
-    Optional<PostgresColumnMetadata> tsColMeta =
-        schemaRegistry.getColumnOrRefresh(tableName, lastUpdatedTsColumn);
-    if (tsColMeta.isEmpty()) {
-      LOGGER.warn(
-          "lastUpdatedTsColumn '{}' not found in schema, skipping timestamp update",
-          lastUpdatedTsColumn);
-      return;
-    }
-
-    long now = System.currentTimeMillis();
-    Object tsValue = convertTimestampForType(now, tsColMeta.get().getPostgresType());
-
-    String sql =
-        String.format(
-            "UPDATE %s SET %s = ? WHERE %s = ?", tableIdentifier, quotedTsColumn, quotedPkColumn);
-
-    try (PreparedStatement ps = connection.prepareStatement(sql)) {
-      for (Key key : keys) {
-        ps.setObject(1, tsValue);
-        ps.setString(2, key.toString());
-        ps.addBatch();
-      }
-      ps.executeBatch();
-    }
-  }
-
-  /** Helper class to hold sub-document update information. */
-  private static class SubDocUpdateEntry {
-
-    final String path;
-    final Document document;
-
-    SubDocUpdateEntry(String path, Document document) {
-      this.path = path;
-      this.document = document;
-    }
+  public BulkUpdateResult bulkUpdateSubDocs(Map<Key, Map<String, Document>> documents) {
+    throw new UnsupportedOperationException(WRITE_NOT_SUPPORTED);
   }
 
   @Override
