updated

Author: Komal Yadav

cdap-metadata-ext-spanner/src/main/java/io/cdap/cdap/metadata/spanner/SpannerMetadataStorage.java

@@ -533,7 +533,6 @@ private SearchResponse performSpannerSearch(SearchRequest request,
       LOG.info("Executing Spanner SQL Template: {}", statement.getSql());
       LOG.info("With Parameters: {}", statement.getParameters());
 
-      // 3. Execute and process results.
       ResultSet resultSet = transaction.executeQuery(statement);
       List<MetadataRecord> results = new ArrayList<>();
       String nextActualCursor = null;
@@ -545,8 +544,7 @@ private SearchResponse performSpannerSearch(SearchRequest request,
 
       LOG.info("Found {} results.", results.size());
 
-      // 4. Create the final response, including the next page's cursor.
-      return createSpannerSearchResponse(request, results, nextActualCursor);
+      return createSearchResponse(request, results, nextActualCursor);
 
     }
   }
@@ -563,18 +561,16 @@ private QueryBuildResult buildSpannerQuery(SearchRequest request, @Nullable Curs
     List<String> sortColumns = sortDetails.getColumns();
     List<Sorting.Order> sortOrders = sortDetails.getOrders();
 
-    // Step 2: Add standard filter conditions (namespaces, types, etc.) to the WHERE clause.
+    // Add standard filter conditions (namespaces, types, etc.) to the WHERE clause.
     appendFilterConditions(request, conditions, params);
 
-    // Step 3: Add the special WHERE condition for keyset pagination if a cursor exists.
+    // Add the special WHERE condition for keyset pagination if a cursor exists.
     appendCursorCondition(requestCursor, sortColumns, sortOrders, conditions, params);
-
-    // Step 4: Assemble the final query string.
     if (!conditions.isEmpty()) {
       sql.append(" WHERE ").append(String.join(" AND ", conditions));
     }
 
-    // Step 5: Add the ORDER BY clause.
+    // Add the ORDER BY clause.
     List<String> orderByClauses = new ArrayList<>();
     for (int i = 0; i < sortColumns.size(); i++) {
       orderByClauses.add(sortColumns.get(i) + " " + sortOrders.get(i).name());
@@ -613,7 +609,6 @@ private SortDetailsResult getSortDetails(SearchRequest request) {
       orders.add(Sorting.Order.ASC);
     }
 
-    // Return the new type-safe object
     return new SortDetailsResult(columns, orders);
   }
 
@@ -719,7 +714,6 @@ private String convertToRegexpPattern(String sqlWildcardPattern) {
   }
 
   /**
-   * UPDATED HELPER METHOD
    * Builds the condition for a simple term search based on the provided scope.
    * Now uses the precise SEARCH_SUBSTRING syntax.
    *
@@ -751,46 +745,59 @@ private String buildScopedSearchCondition(String searchTerm, @Nullable MetadataS
   }
 
   /**
-   * NEW HELPER METHOD
    * Appends the complex WHERE condition for keyset pagination if a cursor is provided.
    * This isolates the most complicated part of the query construction.
    */
   private void appendCursorCondition(@Nullable Cursor requestCursor, List<String> sortColumns,
                                      List<Sorting.Order> sortOrders, List<String> conditions,
                                      Map<String, Value> params) {
-    if (requestCursor == null || requestCursor.getActualCursor() == null || requestCursor.getActualCursor().isEmpty()) {
-      return;
+    // 1. Validate the cursor
+    if (requestCursor == null || requestCursor.getActualCursor() == null) {
+      return; // No cursor, no pagination condition needed.
     }
 
     String[] cursorValues = requestCursor.getActualCursor().split(",", -1);
     if (cursorValues.length != sortColumns.size()) {
-      LOG.warn("Cursor values do not match sort columns. Ignoring cursor. Cursor: '{}', SortColumns: '{}'",
-               requestCursor.getActualCursor(), sortColumns);
-      return;
+      LOG.warn("Cursor values count ({}) does not match sort columns count ({}). Ignoring cursor. " +
+                 "Cursor: '{}', SortColumns: '{}'",
+               cursorValues.length, sortColumns.size(), requestCursor.getActualCursor(), sortColumns);
+      return; // Invalid cursor format, cannot apply pagination.
     }
 
-    List<String> pageConditions = new ArrayList<>();
+    List<String> combinedRowConditions = new ArrayList<>();
+    List<String> prefixEqualityConditions = new ArrayList<>();
+
+    // Iterate through each column in the sort order to build the multi-column comparison
     for (int i = 0; i < sortColumns.size(); i++) {
-      List<String> keyConditions = new ArrayList<>();
-      String colName = sortColumns.get(i);
-
-      for (int j = 0; j < i; j++) {
-        String priorCol = sortColumns.get(j);
-        String paramName = "cursor_" + priorCol;
-        keyConditions.add(String.format("%s = @%s", priorCol, paramName));
-        params.put(paramName, Value.string(cursorValues[j]));
+      String currentSortColumn = sortColumns.get(i);
+      Sorting.Order currentSortOrder = sortOrders.get(i);
+      String currentCursorValue = cursorValues[i];
+
+      // Add the equality condition for the *previous* column to the prefix list
+      if (i > 0) {
+        String priorColumn = sortColumns.get(i - 1);
+        String priorCursorValue = cursorValues[i - 1];
+        String paramName = "cursor_eq_" + priorColumn + "_" + (i - 1);
+        params.put(paramName, Value.string(priorCursorValue));
+        prefixEqualityConditions.add(String.format("%s = @%s", priorColumn, paramName));
       }
 
-      String operator = sortOrders.get(i) == Sorting.Order.ASC ? ">" : "<";
-      String paramName = "cursor_" + colName;
-      keyConditions.add(String.format("%s %s @%s", colName, operator, paramName));
-      params.put(paramName, Value.string(cursorValues[i]));
+      // Start building the conditions for the current OR clause
+      List<String> currentClauseConditions = new ArrayList<>(prefixEqualityConditions);
+      String operator = (currentSortOrder == Sorting.Order.ASC) ? ">" : "<";
+      String paramName = "cursor_gtlt_" + currentSortColumn + "_" + i;
+      params.put(paramName, Value.string(currentCursorValue));
+      currentClauseConditions.add(String.format("%s %s @%s", currentSortColumn, operator, paramName));
+      combinedRowConditions.add("(" + String.join(" AND ", currentClauseConditions) + ")");
+    }
 
-      pageConditions.add("(" + String.join(" AND ", keyConditions) + ")");
+    // 3. Add the final OR-ed condition to the main conditions list
+    if (!combinedRowConditions.isEmpty()) {
+      conditions.add("(" + String.join(" OR ", combinedRowConditions) + ")");
     }
-    conditions.add("(" + String.join(" OR ", pageConditions) + ")");
   }
 
+
   /**
    * Creates the "actualCursor" part of the next cursor. This is the "bookmark" itself.
    * It's now based on the actual columns that were used for sorting.
@@ -807,7 +814,7 @@ private String createNextCursorKey(ResultSet resultSet, List<String> sortColumns
   /**
    * Creates the final SearchResponse, correctly packaging the next cursor string.
    */
-  private SearchResponse createSpannerSearchResponse(SearchRequest request, List<MetadataRecord> results,
+  private SearchResponse createSearchResponse(SearchRequest request, List<MetadataRecord> results,
                                                                                String nextActualCursor) {
     String finalCursorString = (nextActualCursor != null) ?
       getCursor(request, results, nextActualCursor).toString() : null;
@@ -838,12 +845,9 @@ private Cursor getCursor(SearchRequest request, List<MetadataRecord> results, St
   private MetadataRecord mapSpannerResult(ResultSet resultSet) {
     String documentId = resultSet.getString(Tables.Metadata.METADATA_ID_FIELD);
     Struct row = resultSet.getCurrentRowAsStruct();
-    LOG.info(row.toString());
     String metadataJson = row.getJson(7);
     Metadata metadata = GSON.fromJson(metadataJson, Metadata.class);
     MetadataEntity entity = toMetadataEntity(documentId);
-
-
     return new MetadataRecord(entity, metadata);
   }
 
@@ -861,25 +865,25 @@ private static String mapSortKey(String key) {
   }
 
   /**
-   * Translate a document id in the index into a metadata entity.
+   * Translate a metadata id in the index into a metadata entity.
    */
-  private static MetadataEntity toMetadataEntity(String documentId) {
-    int index = documentId.indexOf(':');
+  private static MetadataEntity toMetadataEntity(String metadataId) {
+    int index = metadataId.indexOf(':');
     if (index < 0) {
       throw new IllegalArgumentException(
-        "Document Id must be of the form 'type:k=v,...' but is " + documentId);
+        "Metadata Id must be of the form 'type:k=v,...' but is " + metadataId);
     }
-    String type = documentId.substring(0, index);
+    String type = metadataId.substring(0, index);
     MetadataEntity.Builder builder = MetadataEntity.builder();
-    for (String part : documentId.substring(index + 1).split(",")) {
+    for (String part : metadataId.substring(index + 1).split(",")) {
       String[] parts = part.split("=", 2);
       if (parts[0].equals(type)) {
         builder.appendAsType(parts[0], parts[1]);
       } else {
         builder.append(parts[0], parts[1]);
       }
     }
-    // TODO (CDAP-13597): Handle versioning of metadata entities in a better way
+
     // if it is a versioned entity then add the default version
     return MetadataUtil.addVersionIfNeeded(builder.build());
   }