feat: add trigger_by_flush flag in barrier

[ZENOTME]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

context:
#13899 will caused sink not necessary flush

This PR involve lots interface change so I'm not sure whether it's good solution.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[BugenZhao]

• Any concern for not making this into BarrierKind?
• Previously we had a discussion (feat(meta): add checkpoint_frequency and support decoupling #4966 (comment)) on whether to introduce a CHECKPOINT command aside of FLUSH. I think we're aligning the semantics of FLUSH more closely to what CHECKPOINT represents in Postgres. Do you think it's better to separate them into two commands now and only flush the sink log store on CHECKPOINT?

[liurenjie1024]

cc @wenym1 PTAL

[ZENOTME]

• Previously we had a discussion (feat(meta): add checkpoint_frequency and support decoupling #4966 (comment)) on whether to introduce a CHECKPOINT command aside of FLUSH. I think we're aligning the semantics of FLUSH more closely to what CHECKPOINT represents in Postgres. Do you think it's better to separate them into two commands now and only flush the sink log store on CHECKPOINT?

(Seems I can't find the FLUSH definition of PostgreSQL.

After introducing sink commit decouple, I think there are three semantics:

• Barrier: After the barrier, the internal state will be visible to the user.
• Checkpoint: After the checkpoint, the visible internal state will be flushed. But after decoupling sink commits, the sink data(external data) may not be guaranteed to be visible and flush to the user.
• Flush: basically, it means checkpoint + sink data(external state) visible and flush for the user.

Do you think it's better to separate them into two commands now and only flush the sink log store on CHECKPOINT?

I think we need to have two commands, but I'm concerned whether we should call Flush as Checkpoint. According above definition, we can call them as they are.

• Any concern for not making this into BarrierKind?

From the perspective of implementation, this way may involve less modification because the flush is still a checkpoint. Except for the sink, other parts of the system still treat flush as a checkpoint. Only the sink needs to check whether the checkpoint is triggered by flush. This means that code like if barrier_kind == BarrierKind::Checkpoint doesn't need to modify to if barrier_kind == BarrierKind::Checkpoint || barrier_kind == BarrierKind::Flush, this modification may be easy to miss.

For users, they have different semantics. But I'm concerned about whether we need to treat them as two commands in internal implementation. Because flush means checkpoint + commit sink, in this perspective, in most parts of the system except for the sink, we can treat flush as a "checkpoint".

[BugenZhao]

(Seems I can't find the FLUSH definition of PostgreSQL.

Yes. This is a command made up by ourselves.

After introducing sink commit decouple, I think there are three semantics:

My thoughts is that we will have 2 commands in the future:

• FLUSH: send a non-checkpoint barrier and wait for it to be collected.
• CHECKPOINT: send a checkpoint barrier and wait for it to be collected, also flush log store.
  • Can also make whether to flush the log store as an option for the CHECKPOINT command.

this modification may be easy to miss.

I imagined the opposite. 😂 By adding a new variant to the enum, there'll naturally be some compiling errors indicating that which part of code should be updated. Also developers can grep the codebase for BarrierKind:: to find if there's any == or matches! to be updated.

However, by introducing a new field alongside the existing kind field, the code will compile successfully without any indication of that. The author and the reviewers of this PR may know that the new field only affects the behavior of sinks, but developers in the future do not since it's assigned as a so general name as trigger_by_flush instead of something like should_flush_log_store. Developers may reuse the field and assign it semantics that we don't expect in this PR.

[wenym1]

IIUC, trigger_by_flush should mean flush all unconsumed messages in log store?

If so, on trigger_by_flush, I think we should not serialize and write the barrier to log store storage. Instead we should wait for log reader to finish consuming all messages before return.

[ZENOTME]

IIUC, trigger_by_flush should mean flush all unconsumed messages in log store?

If so, on trigger_by_flush, I think we should not serialize and write the barrier to log store storage. Instead we should wait for log reader to finish consuming all messages before return.

I think it means a checkpoint barrier before we introduce the sink commit decouple.🤔 What I think is to make flush do it like before. I'm confused about consuming all messages, does it mean all messages before the checkpoint barrier?

[ZENOTME]

My thoughts is that we will have 2 commands in the future:

• FLUSH: send a non-checkpoint barrier and wait for it to be collected.
• CHECKPOINT: send a checkpoint barrier and wait for it to be collected, also flush log store.
  • Can also make whether to flush the log store as an option for the CHECKPOINT command.

This looks reasonable to me. I think it can be a separate PR. We can:

1. Let the FLUSH be the CHECKPOINT you mean above.
2. Add FLUSH and make original FLUSH be CHECKPOINT
3. make flush the log store as an option for the CHECKPOINT command.

I imagined the opposite. 😂 By adding a new variant to the enum, there'll naturally be some compiling errors indicating that which part of code should be updated. Also developers can grep the codebase for BarrierKind:: to find if there's any == or matches! to be updated.

However, by introducing a new field alongside the existing kind field, the code will compile successfully without any indication of that. The author and the reviewers of this PR may know that the new field only affects the behavior of sinks, but developers in the future do not since it's assigned as a so general name as trigger_by_flush instead of something like should_flush_log_store. Developers may reuse the field and assign it semantics that we don't expect in this PR.

Cool! Let's make it a BarrierKind::🤣

[wenym1]

IIUC, trigger_by_flush should mean flush all unconsumed messages in log store?

If so, on trigger_by_flush, I think we should not serialize and write the barrier to log store storage. Instead we should wait for log reader to finish consuming all messages before return.

I think it means a checkpoint barrier before we introduce the sink commit decouple.🤔 What I think is to make flush do it like before. I'm confused about consuming all messages, does it mean all messages before the checkpoint barrier?

Yes, it means consume all messages before the barrier for sinks with decouple enabled. I think this is the key difference between the flush command and a normal checkpoint command.

For implementation, there should not be a new message kind in the log store. On flush, the log writer should wait for log reader to consume all the messages written previously.

[ZENOTME]

After discuss with @wenym1, I find that this is issue already introduce after risingwavelabs/rfcs#55.

After enabling a decouple sink, the flush will not be guaranteed to commit sink data. But it will not affect the correctness because:

• if the user didn't enable new decouple features, the flush works like before, for iceberg, it will guarantee to commit sink data.
• if the user enables these new features, the flush will not be guaranteed to commit sink data.

So I think it can be a new feature rather than a necessary fix before #13899.

May be we can track it together with following refine:

• FLUSH: send a non-checkpoint barrier and wait for it to be collected.
• CHECKPOINT: send a checkpoint barrier and wait for it to be collected, also flush log store.
  • Can also make whether to flush the log store as an option for the CHECKPOINT command.

[xxhZs]

Possibly related, adding the ability to flush the logstore may be beneficial to ci, previously clickhouse sink ci failed for a similar reason, which could only be solved by increasing the timeout at that time

[wenym1]

Possibly related, adding the ability to flush the logstore may be beneficial to ci, previously clickhouse sink ci failed for a similar reason, which could only be solved by increasing the timeout at that time

Is there any context to the ci failure? If we manually disable sink decouple by setting set SINK_DECOUPLE = false, there is no need to flush logstore.

[fuyufjh]

After reading the comments above, this seems to be the final decision, right?

• FLUSH: send a non-checkpoint barrier and wait for it to be collected.
• CHECKPOINT: send a checkpoint barrier and wait for it to be collected, also flush log store.
  • Can also make whether to flush the log store as an option for the CHECKPOINT command.

So shall we close this PR? and create a new issue for this idea?

[ZENOTME]

After reading the comments above, this seems to be the final decision, right?

• FLUSH: send a non-checkpoint barrier and wait for it to be collected.

• CHECKPOINT: send a checkpoint barrier and wait for it to be collected, also flush log store.

  • Can also make whether to flush the log store as an option for the CHECKPOINT command.

So shall we close this PR? and create a new issue for this idea?

Yes, I think so.