Event streamer tries to read from dropped chunks #206
Comments
|
Hey @vidosits! Thanks for the report. I think the change is perfectly fine. It'd be awesome if you can send a pull request for it. 🙏 |
|
Thanks, I tried with the test, but I only got like this far before having had to work on work stuff :( timescaledb-event-streamer/internal/systemcatalog/systemcatalog_test.go /*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* 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.
*/
package systemcatalog
import (
"github.com/jackc/pgx/v5"
"github.com/noctarius/timescaledb-event-streamer/spi/config"
"github.com/noctarius/timescaledb-event-streamer/spi/pgtypes"
"github.com/noctarius/timescaledb-event-streamer/spi/systemcatalog"
"github.com/stretchr/testify/assert"
"testing"
)
func Test_Dropped_Chunks_Are_Excluded(
t *testing.T,
) {
var userConfig pgx.ConnConfig
var testConfig config.Config
systemCatalog, err := NewSystemCatalog(&testConfig, &userConfig, nil, nil, nil, nil, nil, nil)
if err != nil {
t.Fatalf("Couldn't create system catalog: %+v", err)
}
for i := 0; i < 10; i++ {
systemcatalog.NewChunk(.. .. .., dropped = True),
}
chunks := systemCatalog.GetAllChunks()
assert.Equal(t, false, chunks == nil)
}
func makeHypertable(
id int32, schemaName, tableName string,
) *systemcatalog.Hypertable {
return systemcatalog.NewHypertable(
id,
schemaName,
tableName,
"test",
"test",
nil,
0,
false,
nil,
nil,
pgtypes.DEFAULT,
)
} |
|
That's fine. I'll add the test. You took the "unit test" a bit strikter than I expected. The are very little real unit tests, since most tests require the database to be available. They're more like integration tests. There's a whole construct around those things. Anyway, for a bit more understanding, it is possible to mock out quite a few things to retrieve data, but it's gonna be complicated and you'd have to mock all of the actual database access layer. It's easier to just build the test like this one. You will have to start and stop the streamer, and you need to make sure you have the state persistence working for the test. But as I said, happy to build the test if you send a PR 👍 |
|
Thanks! I did send #207 . RE: Tests, you're of course completely right, I just didn't know how strictly you meant unit test :) I was gonna go with something like this initally: func (pts *PublicationTestSuite) Test_Dropped_Chunks_Are_Not_Processed() {
testSink := testsupport.NewEventCollectorSink()
publicationName := lo.RandomString(10, lo.LowerCaseLettersCharset)
var tableName string
chunksNotToBeProcessed := make([]string, 0)
pts.RunTest(
func(ctx testrunner.Context) error {
existingChunks, publishedChunks, err := readAllAndPublishedChunks(ctx, tableName, publicationName)
if err != nil {
return err
}
*** somewhere around here we should be testing if the streamer has picked up dropped chunks or not ***
return nil
},
testrunner.WithSetup(func(ctx testrunner.SetupContext) error {
_, tn, err := ctx.CreateHypertable("ts", time.Hour,
testsupport.NewColumn("ts", "timestamptz", false, true, nil),
testsupport.NewColumn("val", "integer", false, false, nil),
)
if err != nil {
return err
}
tableName = tn
if _, err := ctx.Exec(context.Background(),
fmt.Sprintf(
"INSERT INTO \"%s\" SELECT ts, ROW_NUMBER() OVER (ORDER BY ts) AS val FROM GENERATE_SERIES('2023-03-25 00:00:00'::TIMESTAMPTZ, '2023-03-25 23:59:59'::TIMESTAMPTZ, INTERVAL '1 minute') t(ts)",
tableName,
),
); err != nil {
return err
}
if droppedChunks, err := ctx.Query(context.Background(),
fmt.Sprintf(
"SELECT drop_chunks('%s', '2023-03-25 12:00:00'::TIMESTAMPTZ)",
tableName,
),
); err != nil {
return err
} else {
for droppedChunks.Next() {
var chunkName string
if err := droppedChunks.Scan(&chunkName); err != nil {
return nil
}
chunksNotToBeProcessed = append(chunksNotToBeProcessed, chunkName)
}
}
ctx.AddSystemConfigConfigurator(testSink.SystemConfigConfigurator)
ctx.AddSystemConfigConfigurator(func(config *sysconfig.SystemConfig) {
config.PostgreSQL.Publication.Name = publicationName
})
return nil
}),
)
}but then I was like, wait does this go into replicator tests or somewhere else? oh maybe I just just test the func or whatever anyways, I'll take a look once you've done it and then be more productive the next time :) |
|
The looks like a good start. You'd have to get the replicator started once, so that the publication and everything is set up and the persistence state file knows about the existing chunks. At that point you'd suspend the replicator, which will shut it down, you drop the chunk, and resume (or restart) the replicator. That should now fail, do to the state file knowing about a chunk which doesn't exist anymore (or doesn't work anymore). At least that's my guess about what happens, but maybe that a second case which may fail 🫣 |
It seems to me that event streamer is trying to read from / work with dropped hypertable chunks, which of course it can't find.
Checking chunks in timescaledb catalog:
Checking the tables in
_timescaledb_internalschema:My best idea for fixing this would be to:
GetAllChunks()is called heredoes either of these make sense?
I've used the following patch and it worked, but I'm wondering if this is the right thing to do.
The text was updated successfully, but these errors were encountered: