Enable custom query in snapshot mode

[Abdulsametileri]

Some users want to specialize snapshot query. Currently, we are using

func (s *Snapshotter) buildChunkQuery(chunk *Chunk, orderByClause string, pkColumns []string) string {
	if chunk.hasRangeBounds() && len(pkColumns) == 1 {
		pkColumn := pkColumns[0]

		return fmt.Sprintf(
			"SELECT * FROM %s.%s WHERE %s >= %d AND %s <= %d ORDER BY %s LIMIT %d",
			chunk.TableSchema,
			chunk.TableName,
			pkColumn,
			*chunk.RangeStart,
			pkColumn,
			*chunk.RangeEnd,
			orderByClause,
			chunk.ChunkSize,
		)
	}

	return fmt.Sprintf(
		"SELECT * FROM %s.%s ORDER BY %s LIMIT %d OFFSET %d",
		chunk.TableSchema,
		chunk.TableName,
		orderByClause,
		chunk.ChunkSize,
		chunk.ChunkStart,
	)
}

we can get the query field from config.SnapshotConfig and use here.

• SELECT * FROM %s.%s WHERE %s >= %d AND %s <= %d AND {userCondition} ORDER BY %s LIMIT %d
• SELECT * FROM %s.%s WHERE {userCondition} ORDER BY %s LIMIT %d OFFSET %d

we can also get this condition to the getTableRawCount table count request.

func (s *Snapshotter) getTableRawCount(ctx context.Context, schema, table string) (int64, error) {
     query := fmt.Sprintf("SELECT COUNT(*) FROM %s.%s WHERE {userCondition}", schema, table)

[Abdulsametileri]

@ardakyigit will look at this.

[briankski]

👍 for this as well, I need a way to include/exclude specific columns for snapshots as well. I added #95 that should work for the cdc streaming changes.