feat: implement filter in cdc and incremental by vikaxsh · Pull Request #756 · datazip-inc/olake

vikaxsh · 2026-01-14T13:00:12Z

Description

This PR adds filtering capability for CDC and incremental sync modes, allowing users to selectively sync records based on configurable filter conditions.

Fixes # (issue)

Type of change

Multiple Conditions: Support for multiple filter conditions with AND/OR logical operators
Operators: Supports =, !=, <, >, <=, >=
Type-Aware Filtering: Automatic type conversion and validation based on schema

New feature (non-breaking change which adds functionality)

How Has This Been Tested?

Comprehensive test coverage in destination/filter_test.go including:

Legacy filter support
Multiple operators and data types
AND/OR logic
Null value handling
Type coercion
Edge cases
Scenario A
Scenario B

Screenshots or Recordings

Documentation

Documentation Link: [link to README, olake.io/docs, or olake-docs]
N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

New Filter Format

Filters can be specified in the stream configuration using the new FilterInput structure:

{
  "selected_streams": {
    "namespace": [
      {
        "stream_name": "table_1",
        "filter_input": {
          "logical_operator": "AND",
          "conditions": [
            {
              "column": "id",
              "operator": ">",
              "value": 100
            },
            {
              "column": "status",
              "operator": "=",
              "value": "active"
            }
          ]
        }
      }
    ]
  }
}

Legacy Filter Format (Still Supported)

The legacy string-based filter format continues to work:

{
  "filter": "id > 100 AND status = \"active\""
}

…eat/filter-in-cdc&incremental

constants/constants.go

destination/filter.go

destination/parquet/parquet.go

types/catalog.go

types/stream_configured.go

drivers/mongodb/internal/backfill.go

drivers/mongodb/internal/mon_test.go

drivers/mongodb/internal/mon_test_util.go

pkg/jdbc/jdbc.go

vaibhav-datazip · 2026-01-24T13:17:32Z

pkg/jdbc/jdbc.go

+		// ---------- NULL handling ----------
+		if cond.Value == nil {


earlier this check was used to filter out null values like field != null , will it work the same way now as well . is it backward compatible.

i have tested with putting null value in filter as well, worked fine

vaibhav-datazip · 2026-01-24T13:21:43Z

pkg/jdbc/jdbc.go

+			if driverType == constants.Oracle && strings.Contains(v, "T") && (strings.Contains(v, "Z") || strings.Contains(v, "+") || (strings.Contains(v, "-") && len(v) > 19)) {
+				if t, err := time.Parse(time.RFC3339, v); err == nil {
+					timestampStr := t.Format("2006-01-02 15:04:05.000")
+					valueSQL = fmt.Sprintf(
+						"TO_TIMESTAMP('%s', 'YYYY-MM-DD HH24:MI:SS.FF')",
+						timestampStr,
+					)
+				}


earlier the user had to enter the timestamp format in oracle specific timestamp format for the filter to work , what will this work if user enters the timestamp in that format ?

have tested with this format as well 01-JUL-22 03.30.00.000000 PM, will work in Full load but not in cdc as ReformatValue function only support (YYYY-MM-DD) for parsing timestamp from string

so wont the timestamp filter work in cdc ?

it will work only if provided in this format (YYYY-MM-DD)

pkg/jdbc/jdbc.go

drivers/oracle/internal/oracle_test.go

destination/writers.go

protocol/sync.go

destination/writers.go

vaibhav-datazip · 2026-01-26T04:48:48Z

drivers/mysql/internal/mysql_test.go

+		FilterInput: `{
+                    "logical_operator": "And",
+                    "conditions": [
+                        {
+                            "column": "price_double",
+                            "operator": "<",
+                            "value": 239834.89
+                        },
+                        {
+                            "column": "created_timestamp",
+                            "operator": ">=",
+                            "value": "2022-07-01T15:30:00.000+00:00"
+                        }
+                    ]


can we add other datatypes filter as well ?

it would make integration test run longer, all datatypes are already covered in filter_test.go

we can discuss this

…eat/filter-in-cdc&incremental

vishalm0509 · 2026-02-13T02:52:31Z

types/catalog.go

+	//legacy filter input
+	Filter string `json:"filter,omitempty"`
+	//new filter input
+	FilterInput *FilterInput `json:"filter_input,omitempty"`


Should we add this change also to clear-destination check ? To this function GetStreamsDelta()

do u mean if user changes the filter?

Yes, filter is there, but filter_config is not there

vishalm0509 · 2026-02-13T03:30:12Z

drivers/mongodb/internal/backfill.go

 }

-func buildMongoCondition(cond types.Condition) bson.D {
+func buildMongoCondition(cond types.FilterCondition) bson.D {


The SQLFilter now supports multiple filter conditions, but here in Mongodb we are restricting to 2.

Shouldn't it be consistent ?

vishalm0509 · 2026-02-25T10:32:24Z

utils/typeutils/filter.go

+// FilterRecords applies filtering ONLY for new filters.
+// For legacy filters, records are returned unchanged.


Redundant comment

vikaxsh added 12 commits January 14, 2026 18:29

feat: implement filter in cdc

c9dc5d3

feat: add log for flatten record count

b3195f8

fix: github workflows

9cdb42a

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

06be111

…eat/filter-in-cdc&incremental

fix: lint issue

7be9043

fix: lint issue

485cf52

feat: filter only for cdc and backfill

4bc38e9

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

adb97a5

…eat/filter-in-cdc&incremental

fix: integration test for incremental

1dab2d1

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

69dee47

…eat/filter-in-cdc&incremental

feat: add filter in full load for specific drivers

4d681ac

feat: add unit test for filter

ab0848f

vikaxsh force-pushed the feat/filter-in-cdc&incremental branch from b75a508 to ab0848f Compare January 20, 2026 10:26

vikaxsh added 12 commits January 20, 2026 17:02

fix: add mongo integration test

d12ec1b

fix: skip filtring of delete records

b163ac6

fix: add mysql, postgres, oracle integration test

855eae0

fix: integration test

4ec7731

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

4a974c4

…eat/filter-in-cdc&incremental

fix: integration test

94f9d89

fix: integration test for oracle

0f1f35c

fix: integration test for oracle

7da5ccb

fix: integration test for oracle

cfacf76

chore: cleanup

57514bc

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

6d7ed0b

…eat/filter-in-cdc&incremental

fix: lint issue

5af80db

vikaxsh marked this pull request as ready for review January 21, 2026 08:21

Merge branch 'staging' into feat/filter-in-cdc&incremental

91ba0f2

vaibhav-datazip reviewed Jan 23, 2026

View reviewed changes

vaibhav-datazip reviewed Jan 26, 2026

View reviewed changes

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

ea4939e

…eat/filter-in-cdc&incremental

vikaxsh had a problem deploying to integration_tests February 22, 2026 15:03 — with GitHub Actions Error

fix: add filter value parsing in filter

2cc51da

vikaxsh had a problem deploying to integration_tests February 23, 2026 07:33 — with GitHub Actions Error

fix: add filter unit tests again

db18902

vikaxsh requested a deployment to integration_tests February 25, 2026 06:45 — with GitHub Actions Waiting

fix: lint issue

8b16aa0

vikaxsh had a problem deploying to integration_tests February 25, 2026 06:52 — with GitHub Actions Error

chore: change name to filter config

da7cb54

vikaxsh had a problem deploying to integration_tests February 25, 2026 07:05 — with GitHub Actions Error

fix: monog backfill filter

f84574f

vikaxsh had a problem deploying to integration_tests February 25, 2026 09:21 — with GitHub Actions Failure

vikaxsh had a problem deploying to integration_tests February 25, 2026 10:15 — with GitHub Actions Failure

vishalm0509 reviewed Feb 25, 2026

View reviewed changes

fix: mongo incr itegration test

c9d1eb3

vikaxsh had a problem deploying to integration_tests February 25, 2026 13:49 — with GitHub Actions Failure

vikaxsh had a problem deploying to integration_tests February 25, 2026 16:17 — with GitHub Actions Failure

vikaxsh temporarily deployed to integration_tests February 26, 2026 06:41 — with GitHub Actions Inactive

chore: clear destination in change of filter

d6b39cf

vikaxsh requested a deployment to integration_tests February 26, 2026 07:45 — with GitHub Actions Waiting

fix: unit test

bae5499

vikaxsh had a problem deploying to integration_tests February 26, 2026 08:07 — with GitHub Actions Failure

fix: delete integration test

ca51f9d

vikaxsh had a problem deploying to integration_tests February 26, 2026 10:00 — with GitHub Actions Failure

vikaxsh had a problem deploying to integration_tests February 26, 2026 11:56 — with GitHub Actions Failure

fix: delete integration test for parquet

fe3795b

vikaxsh had a problem deploying to integration_tests February 26, 2026 13:48 — with GitHub Actions Failure

fix: delete integration test for parquet

53992f4

vikaxsh temporarily deployed to integration_tests February 26, 2026 15:12 — with GitHub Actions Inactive

Merge branch 'staging' into feat/filter-in-cdc&incremental

2ff69f8

vaibhav-datazip requested a deployment to integration_tests February 27, 2026 09:51 — with GitHub Actions Waiting

		// ---------- NULL handling ----------
		if cond.Value == nil {

		// FilterRecords applies filtering ONLY for new filters.
		// For legacy filters, records are returned unchanged.

Conversation

vikaxsh commented Jan 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

How Has This Been Tested?

Screenshots or Recordings

Documentation

Related PR's (If Any):

New Filter Format

Legacy Filter Format (Still Supported)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vikaxsh Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

vikaxsh commented Jan 14, 2026 •

edited

Loading

vikaxsh Jan 27, 2026 •

edited

Loading