Method paginationKeyColumn

table_schema_cache.go:264–306

(cascadingPaginationColumnConfig *CascadingPaginationColumnConfig)

}

func (t *TableSchema) paginationKeyColumn(cascadingPaginationColumnConfig *CascadingPaginationColumnConfig) (*schema.TableColumn, int, error) {
	var err error
	var paginationKeyColumn *schema.TableColumn
	var paginationKeyIndex int

	if paginationColumn, found := cascadingPaginationColumnConfig.PaginationColumnFor(t.Schema, t.Name); found {
		// Use per-schema, per-table pagination key from config
		paginationKeyColumn, paginationKeyIndex, err = t.findColumnByName(paginationColumn)
	} else if len(t.PKColumns) == 1 {
		// Use Primary Key
		paginationKeyIndex = t.PKColumns[0]
		paginationKeyColumn = &t.Columns[paginationKeyIndex]
	} else if fallbackColumnName, found := cascadingPaginationColumnConfig.FallbackPaginationColumnName(); found {
		// Try fallback from config
		paginationKeyColumn, paginationKeyIndex, err = t.findColumnByName(fallbackColumnName)
	} else {
		// No usable pagination key found
		err = NonExistingPaginationKeyError(t.Schema, t.Name)
	}

	if paginationKeyColumn != nil {
		isNumber := paginationKeyColumn.Type == schema.TYPE_NUMBER || paginationKeyColumn.Type == schema.TYPE_MEDIUM_INT
		isBinary := paginationKeyColumn.Type == schema.TYPE_BINARY ||
			paginationKeyColumn.Type == schema.TYPE_STRING

		if !isNumber && !isBinary {
			return nil, -1, NonNumericPaginationKeyError(t.Schema, t.Name, paginationKeyColumn.Name)
		}

		// For string types (VARCHAR, CHAR), validate that the collation is binary
		// BINARY and VARBINARY types don't have collations and are always binary-safe
		// Related PR comment with integration test proof: https://github.com/Shopify/ghostferry/pull/417#discussion_r2619684805
		if paginationKeyColumn.Type == schema.TYPE_STRING && paginationKeyColumn.Collation != "" {
			// Binary collations end with "_bin" (e.g., utf8mb4_bin, latin1_bin)
			// BINARY type has empty collation and is handled above
			if !strings.HasSuffix(paginationKeyColumn.Collation, "_bin") {
				return nil, -1, NonBinaryCollationError(t.Schema, t.Name, paginationKeyColumn.Name, paginationKeyColumn.Collation)
			}
		}
	}

	return paginationKeyColumn, paginationKeyIndex, err
}

// GetPaginationColumn retrieves PaginationKeyColumn
func (t *TableSchema) GetPaginationColumn() *schema.TableColumn {