Skip to content

Commit f16b303

Browse files
authored
Merge branch 'main' into fix/region-termination-bounded-retry
2 parents fe13ee3 + 8803d08 commit f16b303

48 files changed

Lines changed: 2536 additions & 147 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.github/workflows/comment-commands.yml

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -153,7 +153,15 @@ jobs:
153153
reviewers.push(h);
154154
}
155155
if (!reviewers.length && !team_reviewers.length) {
156-
core.warning(`No valid @mentions in '${action}'; skipping.`);
156+
// Explicit @mentions are required — the suggest-reviewers CI job
157+
// already posted a suggestion comment on this PR, so the author
158+
// has candidates to choose from.
159+
await github.rest.issues.createComment({
160+
owner, repo, issue_number: pull_number,
161+
body:
162+
`Please specify at least one reviewer: \`/${action} @user\`.\n` +
163+
`Check the suggestion comment on this PR for candidates.`,
164+
});
157165
return;
158166
}
159167
Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
# On PR open/update, runs `git blame` on every changed file and posts (or
18+
# edits) a single reviewer-suggestion comment split into two buckets:
19+
# • Committers — collaborators who can be formally review-requested.
20+
# The author uses `/request-review @login` to act on these.
21+
# • Non-committer contributors — have context but cannot be review-
22+
# requested; the author can @-mention them to notify.
23+
# The CI never sends a review request on its own.
24+
name: Suggest reviewers
25+
on:
26+
pull_request_target:
27+
types: [opened, synchronize, reopened]
28+
29+
permissions:
30+
contents: read
31+
issues: write
32+
pull-requests: write
33+
34+
jobs:
35+
suggest-reviewers:
36+
runs-on: ubuntu-latest
37+
steps:
38+
- uses: actions/checkout@v5
39+
with:
40+
fetch-depth: 0
41+
- uses: actions/github-script@v8
42+
with:
43+
github-token: ${{ secrets.GITHUB_TOKEN }}
44+
script: |
45+
const { execFileSync } = require('node:child_process');
46+
const pull_number = context.payload.pull_request.number;
47+
const author = context.payload.pull_request.user.login;
48+
const { owner, repo } = context.repo;
49+
50+
const { data: pull } = await github.rest.pulls.get({ owner, repo, pull_number });
51+
52+
try {
53+
execFileSync('git', ['fetch', 'origin', pull.base.ref], { encoding: 'utf8' });
54+
} catch (e) {
55+
core.warning(`git fetch for base ref ${pull.base.ref} failed: ${e.message}`);
56+
}
57+
58+
const files = await github.paginate(github.rest.pulls.listFiles, {
59+
owner, repo, pull_number, per_page: 100,
60+
});
61+
62+
// Parse `git blame -p` output to find the most-recent commit per file.
63+
function latestBlameCommit(blameOutput) {
64+
let latest = null;
65+
let current = null;
66+
67+
function finalizeCurrent() {
68+
if (!current || current.authorTime == null) return;
69+
if (!latest || current.authorTime > latest.authorTime) latest = current;
70+
}
71+
72+
for (const line of blameOutput.split(/\r?\n/)) {
73+
const header = line.match(/^([0-9a-f^]+)\s+\d+\s+\d+\s+\d+$/);
74+
if (header) {
75+
finalizeCurrent();
76+
current = { sha: header[1].replace(/^\^/, ''), authorTime: null };
77+
continue;
78+
}
79+
const authorTime = line.match(/^author-time\s+(\d+)$/);
80+
if (authorTime && current) current.authorTime = Number(authorTime[1]);
81+
}
82+
83+
finalizeCurrent();
84+
return latest;
85+
}
86+
87+
// Count changed files touched per login; track collaborator status.
88+
const committerCounts = new Map(); // collaborators
89+
const nonCommitterCounts = new Map(); // non-collaborators with a GitHub login
90+
91+
for (const { filename, status, previous_filename } of files) {
92+
if (status === 'removed' || status === 'added') continue;
93+
const blamePath = status === 'renamed' ? previous_filename : filename;
94+
95+
let blameOutput;
96+
try {
97+
blameOutput = execFileSync(
98+
'git', ['blame', '-p', pull.base.sha, '--', blamePath],
99+
{ encoding: 'utf8' },
100+
);
101+
} catch (e) {
102+
core.warning(`git blame on ${filename} failed: ${e.message}`);
103+
continue;
104+
}
105+
106+
const latest = latestBlameCommit(blameOutput);
107+
if (!latest) continue;
108+
109+
let commit;
110+
try {
111+
({ data: commit } = await github.rest.repos.getCommit({ owner, repo, ref: latest.sha }));
112+
} catch (e) {
113+
core.warning(`Commit lookup for ${latest.sha} failed: ${e.message}`);
114+
continue;
115+
}
116+
117+
const login = commit.author?.login ?? commit.committer?.login;
118+
if (!login) continue;
119+
if (login.toLowerCase() === author.toLowerCase()) continue;
120+
121+
const loginSource = commit.author?.login ? commit.author : commit.committer;
122+
if (loginSource?.type === 'Bot') continue;
123+
124+
let isCollaborator = false;
125+
try {
126+
await github.rest.repos.checkCollaborator({ owner, repo, username: login });
127+
isCollaborator = true;
128+
} catch (_) { /* not a collaborator */ }
129+
130+
if (isCollaborator) {
131+
committerCounts.set(login, (committerCounts.get(login) ?? 0) + 1);
132+
} else {
133+
nonCommitterCounts.set(login, (nonCommitterCounts.get(login) ?? 0) + 1);
134+
}
135+
}
136+
137+
const MAX_EACH = 3;
138+
139+
const committers = [...committerCounts.entries()]
140+
.sort((a, b) => b[1] - a[1])
141+
.slice(0, MAX_EACH)
142+
.map(([l]) => l);
143+
144+
const nonCommitters = [...nonCommitterCounts.entries()]
145+
.sort((a, b) => b[1] - a[1])
146+
.slice(0, MAX_EACH)
147+
.map(([l]) => l);
148+
149+
const MARKER = '<!-- texera-reviewer-suggestion -->';
150+
151+
let body = `${MARKER}\n`;
152+
body += `### Automated Reviewer Suggestions\n\n`;
153+
body += `Based on the \`git blame\` history of the changed files, we recommend the following reviewers:\n\n`;
154+
155+
if (committers.length) {
156+
body += `- **Committers with relevant context:** ${committers.map(l => `\`@${l}\``).join(', ')}\n`;
157+
body += ` You can request their reviews formally with \`/request-review ${committers.map(l => `@${l}`).join(' ')}\`.\n\n`;
158+
}
159+
160+
if (nonCommitters.length) {
161+
body += `- **Contributors with relevant context:** ${nonCommitters.map(l => `\`@${l}\``).join(', ')}\n`;
162+
body += ` You can notify them by mentioning ${nonCommitters.map(l => `\`@${l}\``).join(', ')} in a comment.\n`;
163+
}
164+
165+
if (!committers.length && !nonCommitters.length) {
166+
body += `- No candidates found from \`git blame\` history.\n`;
167+
}
168+
169+
// Update existing suggestion comment rather than posting a new one.
170+
const allComments = await github.paginate(github.rest.issues.listComments, {
171+
owner, repo, issue_number: pull_number, per_page: 100,
172+
});
173+
const existing = allComments.find(c => c.body?.includes(MARKER));
174+
175+
if (existing) {
176+
await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body });
177+
core.info(`Updated reviewer suggestion comment on #${pull_number}`);
178+
} else {
179+
await github.rest.issues.createComment({ owner, repo, issue_number: pull_number, body });
180+
core.info(`Posted reviewer suggestion comment on #${pull_number}`);
181+
}

amber/LICENSE-binary-python

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -228,7 +228,7 @@ Python packages:
228228
- pympler==1.1
229229
- python-dateutil==2.8.2
230230
- regex==2026.5.9
231-
- requests==2.34.2
231+
- requests==2.34.0
232232
- s3transfer==0.14.0
233233
- safetensors==0.8.0
234234
- tenacity==8.5.0

amber/requirements.txt

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,3 +43,5 @@ SQLAlchemy==2.0.37
4343
pg8000==1.31.5
4444
pympler==1.1
4545
boto3==1.40.53
46+
requests==2.34.0
47+
urllib3==2.7.0

amber/src/main/python/core/runnables/data_processor.py

Lines changed: 5 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -15,26 +15,20 @@
1515
# specific language governing permissions and limitations
1616
# under the License.
1717

18-
import os
1918
import sys
20-
import traceback
2119
from contextlib import contextmanager
2220
from loguru import logger
2321
from threading import Event
2422
from typing import Iterator, Optional
2523

2624
from core.architecture.managers import Context
27-
from core.models import ExceptionInfo, State, TupleLike, InternalMarker
25+
from core.models import State, TupleLike, InternalMarker
2826
from core.models.internal_marker import StartChannel, EndChannel
2927
from core.models.table import all_output_to_tuple
3028
from core.util import Stoppable
29+
from core.util.console_message.error_message import create_error_console_message
3130
from core.util.console_message.replace_print import replace_print
32-
from core.util.console_message.timestamp import current_time_in_local_timezone
3331
from core.util.runnable import Runnable
34-
from proto.org.apache.texera.amber.engine.architecture.rpc import (
35-
ConsoleMessage,
36-
ConsoleMessageType,
37-
)
3832

3933

4034
class DataProcessor(Runnable, Stoppable):
@@ -126,7 +120,9 @@ def _executor_session(self):
126120
logger.exception(err)
127121
exc_info = sys.exc_info()
128122
self._context.exception_manager.set_exception_info(exc_info)
129-
self._report_exception(exc_info)
123+
self._context.console_message_manager.put_message(
124+
create_error_console_message(self._context.worker_id, exc_info)
125+
)
130126
finally:
131127
self._switch_context()
132128

@@ -175,25 +171,5 @@ def _check_and_process_debug_command(self) -> None:
175171
# This line has no side effects on the current debugger state.
176172
self._context.debug_manager.debugger.set_trace()
177173

178-
def _report_exception(self, exc_info: ExceptionInfo):
179-
tb = traceback.extract_tb(exc_info[2])
180-
filename, line_number, func_name, text = tb[-1]
181-
base_name = os.path.basename(filename)
182-
module_name, _ = os.path.splitext(base_name)
183-
formatted_exception = traceback.format_exception(*exc_info)
184-
title: str = formatted_exception[-1].strip()
185-
message: str = "\n".join(formatted_exception)
186-
187-
self._context.console_message_manager.put_message(
188-
ConsoleMessage(
189-
worker_id=self._context.worker_id,
190-
timestamp=current_time_in_local_timezone(),
191-
msg_type=ConsoleMessageType.ERROR,
192-
source=f"{module_name}:{func_name}:{line_number}",
193-
title=title,
194-
message=message,
195-
)
196-
)
197-
198174
def stop(self):
199175
self._running.clear()
Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
import os
19+
import traceback
20+
21+
from core.models import ExceptionInfo
22+
from core.util.console_message.timestamp import current_time_in_local_timezone
23+
from proto.org.apache.texera.amber.engine.architecture.rpc import (
24+
ConsoleMessage,
25+
ConsoleMessageType,
26+
)
27+
28+
29+
def create_error_console_message(
30+
worker_id: str, exc_info: ExceptionInfo
31+
) -> ConsoleMessage:
32+
"""Build an ERROR ``ConsoleMessage`` describing ``exc_info``.
33+
34+
Produces the operator-facing error message for an uncaught exception,
35+
whether it surfaced from a UDF on the data path (DataProcessor) or from a
36+
user expression evaluated on the main loop thread. Sharing this factory
37+
keeps every uncaught-exception path reporting identically; callers are
38+
responsible for recording the exception with the exception manager,
39+
queueing the returned message, and flushing/pausing as appropriate.
40+
"""
41+
tb = traceback.extract_tb(exc_info[2])
42+
if tb:
43+
filename, line_number, func_name, _ = tb[-1]
44+
module_name, _ = os.path.splitext(os.path.basename(filename))
45+
source = f"{module_name}:{func_name}:{line_number}"
46+
else:
47+
# No traceback frames (e.g. an exception object that was never raised).
48+
# Still report it -- an error reporter must not itself throw.
49+
source = ""
50+
formatted_exception = traceback.format_exception(*exc_info)
51+
title: str = formatted_exception[-1].strip()
52+
message: str = "\n".join(formatted_exception)
53+
54+
return ConsoleMessage(
55+
worker_id=worker_id,
56+
timestamp=current_time_in_local_timezone(),
57+
msg_type=ConsoleMessageType.ERROR,
58+
source=source,
59+
title=title,
60+
message=message,
61+
)

0 commit comments

Comments
 (0)