Repository: risingwavelabs/risingwave
File: src/frontend/src/catalog/system_catalog/rw_catalog/rw_fragment_backfill_progress.rs (Rust)

Code before:
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use risingwave_common::types::Fields;
use risingwave_frontend_macro::system_catalog;

/// Provides fragment level backfill progress.
// source backfill progress has the following schema:
// `{"num_consumed_rows": 530, "state": {"Backfilling": "529"}, "target_offset": "9999"}`
#[system_catalog(
view,
"rw_catalog.rw_fragment_backfill_progress",
"with
table_backfill_progress as (select
  progress.job_id,
  progress.fragment_id,
  scan_info.backfill_target_relation_id,
  case
    when scan_info.backfill_type = 'SNAPSHOT_BACKFILL' AND progress.min_epoch > scan_info.backfill_epoch
      then concat('100% (', stats.total_key_count, '/', stats.total_key_count, ')')
    when stats.total_key_count is null
      then '100% (0/0)'
  else
    concat(
      coalesce(progress.current_row_count::numeric / stats.total_key_count::numeric * 100, 0),
      '%',
      ' ',
      '(',
      coalesce(progress.current_row_count, 0),
      '/',
      stats.total_key_count,
      ')'
    )
  end as progress
FROM internal_backfill_progress() progress
JOIN rw_backfill_info scan_info ON progress.job_id = scan_info.job_id AND progress.fragment_id = scan_info.fragment_id
LEFT JOIN rw_table_stats stats ON scan_info.backfill_target_relation_id = stats.id
),
aggregated_source_backfill_progress as (
  select
    source_backfill_progress.job_id,
    source_backfill_progress.fragment_id,
    sum((backfill_progress -> 'num_consumed_rows')::int) as backfill_progress
  from
    internal_source_backfill_progress() source_backfill_progress

Review comment (suggestion):
When table is empty, `rw_table_Stats` will not have an entry. We handle that case here.

Code after:
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use risingwave_common::types::Fields;
use risingwave_frontend_macro::system_catalog;

/// Provides fragment level backfill progress.
// source backfill progress has the following schema:
// `{"num_consumed_rows": 530, "state": {"Backfilling": "529"}, "target_offset": "9999"}`
#[system_catalog(
view,
"rw_catalog.rw_fragment_backfill_progress",
"with
table_backfill_progress as (select
  progress.job_id,
  progress.fragment_id,
  scan_info.backfill_target_relation_id,
  case
    when stats.total_key_count is null
      then '100% (0/0)'
    when scan_info.backfill_type = 'SNAPSHOT_BACKFILL' AND progress.min_epoch > scan_info.backfill_epoch
      then concat('100% (', stats.total_key_count, '/', stats.total_key_count, ')')
  else
    concat(
      coalesce(progress.current_row_count::numeric / stats.total_key_count::numeric * 100, 0),
      '%',
      ' ',
      '(',
      coalesce(progress.current_row_count, 0),
      '/',
      stats.total_key_count,
      ')'
    )
  end as progress
FROM internal_backfill_progress() progress
JOIN rw_backfill_info scan_info ON progress.job_id = scan_info.job_id AND progress.fragment_id = scan_info.fragment_id
LEFT JOIN rw_table_stats stats ON scan_info.backfill_target_relation_id = stats.id
),
aggregated_source_backfill_progress as (
  select
    source_backfill_progress.job_id,
    source_backfill_progress.fragment_id,
    sum((backfill_progress -> 'num_consumed_rows')::int) as backfill_progress
  from
    internal_source_backfill_progress() source_backfill_progress